Ironflow is pre-1.0: APIs can change between releases. Pin a version. What is ready →

@ironflow/node

Node.js SDK for Ironflow. Provides workers, serve handlers, step execution, and a client for server interactions.

For equivalent Tier-1 capability examples, see the Go SDK and Python SDK references.

Installation

The package is public on npm and installs without authentication.

npm install @ironflow/node

The package has seven entry points. @ironflow/node exports serve, createWorker, the client and everything else except the entry points below. @ironflow/node/serve and @ironflow/node/worker export serve (with createHandler) and createWorker, plus related types and helpers. @ironflow/node/worker-streaming (createStreamingWorker) and @ironflow/node/consumer-group-stream (joinConsumerGroupStream) load the protobuf and ConnectRPC packages, which the root entry leaves out. @ironflow/node/test (createTestClient) and @ironflow/node/agent (agent helpers) hold the test client and the agent helpers.

Quick Start

Define a Function

import { ironflow } from '@ironflow/node';

const processOrder = ironflow.createFunction(
  {
    id: 'process-order',
    triggers: [{ event: 'order.placed' }],
  },
  async ({ event, step }) => {
    // Run a step (automatically memoized)
    const validated = await step.run('validate', async () => {
      return validateOrder(event.data);
    });

    // Sleep (durable - survives restarts)
    await step.sleep('wait-for-inventory', '5m');

    // Wait for external event
    const approval = await step.waitForEvent('wait-approval', {
      event: 'order.approved',
      match: 'data.orderId', // JSON path for correlating events
      timeout: '24h',
    });

    // Process payment
    const result = await step.run('process-payment', async () => {
      return processPayment(validated, approval);
    });

    return { success: true, receiptId: result.id };
  }
);

Push Mode (Serverless)

For serverless environments (Next.js, Vercel, AWS Lambda):

// app/api/ironflow/route.ts (Next.js App Router)
import { serve } from '@ironflow/node';

export const POST = serve({
  functions: [processOrder],
  signingKey: process.env.IRONFLOW_SIGNING_KEY,
});

Pull Mode (Worker)

For long-running workers with no timeout limits:

import { createWorker } from '@ironflow/node';

const worker = createWorker({
  serverUrl: 'http://localhost:9123',
  functions: [processOrder],
  maxConcurrentJobs: 10,
});

// Start the worker (blocks until stopped). It drains on SIGINT and SIGTERM,
// then the process exits — see Shutdown under Worker.
await worker.start();

Function Definition

createFunction

Create an Ironflow function.

const myFunction = ironflow.createFunction(
  {
    // Required: unique function ID
    id: 'my-function',

    // Optional: display name
    name: 'My Function',

    // Optional: Zod schema. Enforced at runtime: the handler receives the
    // parsed output; a mismatch fails the run (VALIDATION_ERROR, non-retryable)
    schema: z.object({ userId: z.string() }),

    // Required: event triggers (an empty array for invoke-only functions)
    triggers: [
      { event: 'user.created' },
      { event: 'order.*' },
      { event: 'daily.cleanup', cron: '0 0 * * *' },  // Daily at midnight UTC (`event` is required)
    ],

    // Optional: retry configuration. Only the fields you set are sent; the
    // server defaults are 25 attempts, 1s, 2.0, 5m and onExhausted "pause".
    retry: {
      maxAttempts: 3,
      initialDelayMs: 1000,
      backoffFactor: 2,
      maxDelayMs: 60000,
      onExhausted: 'fail', // 'pause' holds the run for resume, 'fail' ends it
    },

    // Optional: circuit breaker overrides (server defaults: 5, 60s, 10m)
    circuitBreaker: { failureThreshold: 3, openDurationMs: 30000, maxOpenDurationMs: 300000 },

    // Optional: soft start-rate limit; runs over the limit wait in the queue
    throttle: { limit: 100, periodMs: 60000, key: 'customerId' },

    // Optional: execution timeout in milliseconds (unset: no per-function budget)
    timeout: 1800000,

    // Optional: concurrency control
    concurrency: { limit: 10, key: 'customerId' },

    // Optional: declarative cancel-on-event
    // Auto-cancels in-flight runs when a matching event arrives.
    // OR semantics across specs. Tenant-isolated by env_id.
    // See how-to: ../../how-to-guides/workflows/cancel-on-event.md
    cancelOn: [
      { event: 'order.cancelled', match: 'orderId' },
    ],
  },
  async ({ event, step, run, logger }) => {
    // Function implementation
    return { result: 'success' };
  }
);

The config also accepts description, mode, actorKey, secrets, stepTimeout, debounce, recording, recordingProfile and metadata. See FunctionConfig in the core package reference.

Retry settings you leave out are not sent. The server applies its own defaults when each run is created: 25 attempts, 1 second initial delay, a backoff factor of 2.0, a 5 minute maximum delay, and onExhausted: "pause", so a run that uses every attempt is held for ironflow run resume instead of failing. Throw RetryAfterError(message, retryAfterMs) from a handler or a step.run() body to choose the delay before the next attempt. Handlers can read run.attempt, run.maxAttempts and run.onExhausted. With onExhausted: 'fail', a retryable error on the last attempt is terminal and runs the registered compensations; with 'pause' it does not. See Retry, exhaustion and RetryAfterError.

A function ID must not contain the NATS wildcards * or >, whitespace, a control character, or an empty . segment (a..b, .a, a.), and must be under 255 bytes counting the system.function.{id}.registered subject the server builds from it (so 228 bytes of ID). Each returns CodeInvalidArgument / 400 at registration. Dots are allowed — orders.process is a valid ID — and so are non-ASCII IDs.

The rule applies to new registrations only. A function already in the database keeps its ID and still re-registers, so upgrading does not stop a worker that holds a pre-existing malformed ID; its system.function.{id}.* and system.circuit_breaker.{state}.{id} frames are dropped instead, counted by ironflow_system_event_drops_total and logged at warn.

The thrown error carries the server’s reason, not just the status — a rejected ID reads Failed to register function my fn: 400: function ID has an empty segment, whitespace, or a control character. The same applies to worker registration and job polling. The body is truncated at 200 characters, since a proxy in front of the server may return an HTML page rather than JSON.

Function Context

The handler receives a context object:

interface FunctionContext {
  // Triggering event
  event: {
    id: string;
    name: string;
    version: number;
    data: unknown;
    timestamp: Date;
    idempotencyKey?: string;
    source?: string;
    metadata?: Record<string, unknown>;
    emittedBy?: string;
  };

  // Step client for recorded execution
  step: StepClient;

  // Run information
  run: {
    id: string;
    functionId: string;
    attempt: number;
    maxAttempts?: number; // 0 when the engine is too old to send it
    onExhausted?: 'pause' | 'fail' | ''; // empty when the engine is too old to send it
    startedAt: Date;
    environment?: string; // environment of the run's outbound calls
  };

  // Logger instance
  logger: Logger;

  // Resolved environment secrets (read-only)
  secrets: SecretsClient;
}

Step Primitives

run

Execute a step. Results are memoized—if the function restarts, completed steps are skipped.

const result = await step.run('step-id', async () => {
  // This only runs once, even if function retries
  return await someApiCall();
});

// Optional: per-step timeout ("30s", "5m", "1h"); overrides the function's stepTimeout
await step.run('slow-step', () => slowCall(), { timeout: '5m' });

sleep

Sleep for a duration. Durable—survives function restarts.

// Sleep for 5 minutes
await step.sleep('wait', '5m');

// Supported formats: '30s', '5m', '2h', '1d'

sleepUntil

Sleep until a specific time.

await step.sleepUntil('wait-until', new Date('2024-12-25T00:00:00Z'));

waitForEvent

Wait for an external event. The optional payload is stored as the waiting step’s input, available through getRunSteps and the dashboard. It does not affect event matching or the returned event.

const approval = await step.waitForEvent('wait-approval', {
  // Event name pattern
  event: 'order.approved',

  // Optional: JSON path for correlating events (e.g. "data.orderId")
  match: 'data.orderId',

  // Optional: timeout (default: '7d')
  timeout: '24h',
});

match is a path inside event.data. The data. prefix is optional. By default the engine reads the value at match from the event that started the run, and waits for an event with the same value at that path. To correlate on a value that your handler computed, set matchValue to a non-empty literal. The engine then uses it instead of reading the starting event. matchValue requires match. By default, waitForEvent returns the matching event and a deadline fails the run without resuming the handler. Set onTimeout: "return-null" to resume with a memoized null instead. The default type stays Promise<IronflowEvent<T>>; the opt-in type is Promise<IronflowEvent<T> | null>. The policy survives restarts. approve() keeps its fail-on-timeout default.

Yielding steps (sleep, sleepUntil, waitForEvent, invoke, invokeAsync) must propagate their internal signal. If a handler or parallel branch swallows it and returns or attempts another durable operation, the SDK fails nonretryably with an actionable error, including under allSettled. Catch ordinary errors inside the step.run callback. The guard cannot prevent arbitrary side effects performed directly inside the catch block before control returns to the SDK.

Function trigger events, getEvent(), streams.read(), and waitForEvent results expose emittedBy. It is the authenticated emitter that the engine recorded, for example apikey:<id> or user:<id>. The engine sets it from the credential of the caller, so the event body cannot change it. It is undefined when the engine reports none.

parallel

Execute multiple branches in parallel.

const [user, orders, payments] = await step.parallel(
  'fetch-all',
  [
    (step) => step.run('fetch-user', () => fetchUser(userId)),
    (step) => step.run('fetch-orders', () => fetchOrders(userId)),
    (step) => step.run('fetch-payments', () => fetchPayments(userId)),
  ],
  { concurrency: 2, onError: 'failFast' }  // Optional
);

The step each branch receives is load-bearing, not stylistic. This applies to map below as well. A branch is not itself a recorded step — parallel and map only call your callback. Work done with the scoped client is memoized under that branch’s scope; work done directly in the callback body persists nothing and re-runs in full on every retry.

Reaching for the enclosing function’s step instead is a third case: those branches do record real steps, but at the function’s top level rather than under the parallel — a flat timeline, and IDs that stay unique only as long as the branch names differ.

Both forms typecheck, so the SDK logs a warning when every branch in one call skipped the scoped client — once per distinct step name per run. Treat it as a best-effort lint, not a guarantee: it stays silent when any branch throws or yields, and when only some branches skip. For a fan-out that genuinely has nothing to memoize (a pure in-memory transform run through map only for its concurrency limit), pass { expectScopedClient: false } to silence it.

Draining before you migrate. Switching an existing branch from the enclosing step to the scoped one changes its step ID from {runId}:{name}:0 to {runId}:{parallelName}:{branchIndex}:{name}:0. Nothing bridges the two ids, so an in-flight run that resumes after the deploy — one sleeping, waiting for an event, or on a later attempt — finds no memoized row and re-executes those branch steps for real. Let in-flight runs drain before shipping the rewrite.

map

Map over items with automatic parallelization.

const results = await step.map(
  'process-items',
  items,
  async (item, step, index) => {
    return await step.run(`process-${index}`, () => processItem(item));
  },
  { concurrency: 5 }  // Optional: limit parallel execution
);

compensate

Register a compensation handler for saga rollbacks. If the function fails after this step, the compensation runs automatically.

await step.run('reserve-inventory', async () => {
  await reserveInventory(orderId);
});

step.compensate('reserve-inventory', async () => {
  await releaseInventory(orderId);
});

invoke

Synchronously invoke another function and wait for its result.

const result = await step.invoke<PaymentResult>('process-payment', {
  orderId,
  amount,
}, { timeout: '30s' });

invokeAsync

Invoke another function asynchronously. Returns the run ID without waiting for completion.

invokeAsync is already durable. Call it directly, outside step.run. On replay, a completed wrapper skips the callback and its invocation counter. Later calls can then return an earlier child’s run ID without starting a new child. This also applies when calling spawn({ await: false }) inside step.run.

const { runId } = await step.invokeAsync('send-notifications', {
  userId,
  message,
});

For fan-out, call invokeAsync directly on every loop iteration:

const runIds: string[] = [];
for (const cluster of clusters) {
  const { runId } = await step.invokeAsync('open-case', { cluster });
  runIds.push(runId);
}

When upgrading from an SDK that wrapped fire-and-forget spawn calls internally, finish or stop affected parent runs with the old worker before switching versions. Removing the wrapper changes replay counter advancement. Existing partial fan-outs need their child runs reconciled before retrying work.

publish

Publish a message to a topic.

const result = await step.publish('order.events', {
  type: 'order.shipped',
  orderId,
});

options takes an idempotencyKey for server-side deduplication: step.publish(topic, data, { idempotencyKey }). Without it, the step sends a default key derived from its step ID, so a step retry does not publish twice. The server keeps the key for IRONFLOW_RUN_RETENTION_DAYS days (forever when unset or 0) and returns the original event ID on a repeat. One rare gap: if the server stores the message but cannot record it, a retry after 2 minutes stores a second copy under the same event ID. The result has eventId and sequence.

See the @ironflow/node README in the SDK package for detailed step primitive documentation.

Event Data

Parsing Event Data

async ({ event, step }) => {
  // event.data is typed as unknown — cast to your type
  const order = event.data as OrderData;

  // Access raw event fields
  const eventName = event.name;       // e.g. "order.placed"
  const eventId = event.id;
  const eventVersion = event.version; // Schema version (default: 1)

  // Access user-defined metadata attached when the event was emitted
  const traceId = event.metadata?.traceId as string | undefined;
  const source = event.metadata?.source as string | undefined;

  return { orderId: order.id };
}

Client

createClient

Create a client for server interactions:

import { createClient } from '@ironflow/node';

const client = createClient({
  serverUrl: 'http://localhost:9123',  // default: IRONFLOW_SERVER_URL or http://localhost:9123
  apiKey: 'optional-api-key',          // for authenticated servers
  environment: 'env_default_staging',   // X-Ironflow-Environment-ID on REST/Connect calls, not KV, config or secrets (optional; IRONFLOW_ENV is not read)
  timeout: 30000,                       // request timeout in ms (default: 30000)
  onError: (error, context) => {        // global error handler (optional)
    console.error(`[${context.method}] ${error.message}`, context);
  },
  retry: { maxAttempts: 3 },            // retry with backoff (optional, on by default)
});

The createClient() function is the recommended way to create an Ironflow client instance.

Every HTTP request from the Node SDK carries X-Ironflow-SDK: node/<version>, and so do the WebSocket handshakes of the KV watch, config watch and subscription clients. The server does not read it. When a newer SDK calls an older server that lacks the call, the SDK throws a non-retryable IronflowError with code SERVER_TOO_OLD that names the first server version that has it. Two calls do this: the consumer group stream and the projection runner’s lease (AcquireProjectionLease). The error is a ServerTooOldError. Each package entry bundles its own copy of the class, so match err.code === "SERVER_TOO_OLD" rather than instanceof.

Global Error Handler (onError)

The onError callback fires on every client error before the error is re-thrown. It provides centralized error observation for logging, metrics, and alerting without interfering with normal error propagation.

import { createClient, type ErrorContext, type OnErrorHandler } from '@ironflow/node';

const onError: OnErrorHandler = async (error, context) => {
  // Log to your observability stack
  await metrics.increment('ironflow.client.error', {
    method: context.method,
    statusCode: String(context.statusCode ?? 'network'),
  });

  logger.error('Ironflow client error', {
    method: context.method,      // e.g. "emit", "streams.append", "apiKeys.get"
    endpoint: context.endpoint,  // e.g. "/ironflow.v1.IronflowService/Trigger"
    statusCode: context.statusCode, // HTTP status or undefined for network errors
    error: error.message,
  });
};

const client = createClient({ onError });

ErrorContext fields:

Field	Type	Description
`method`	`string`	Client method name (e.g. `"emit"`, `"streams.append"`, `"apiKeys.get"`)
`endpoint`	`string`	ConnectRPC or REST endpoint path
`statusCode`	`number \| undefined`	HTTP status code (`undefined` for network/timeout errors)

Behavior:

The callback is awaited before the error is re-thrown
If the callback throws, the callback error is swallowed (logged to stderr) and the original error is still thrown
Both sync and async callbacks are supported (() => void | Promise<void>)
Propagates to sub-clients: client.kv() and client.config() inherit the handler

Method naming convention: Top-level methods use their name directly (e.g. "emit", "health"). Namespaced methods use dot notation (e.g. "streams.append", "apiKeys.create", "kv.bucket.get", "config.set").

Retry and Backoff (retry)

The client retries a failed request with backoff. Retry is on by default. The defaults are the same as in the Go SDK.

const client = createClient({
  retry: {
    maxAttempts: 3,                // total attempts, the first one included (default: 3)
    initialDelayMs: 100,           // delay before the first retry (default: 100)
    backoffMultiplier: 2,          // the delay grows by this factor (default: 2)
    maxDelayMs: 10000,             // upper limit for one delay (default: 10000)
    connectionRetryDelayMs: 2000,  // fixed delay after a network error or timeout (default: 2000)
    onRetry: ({ attempt, maxAttempts, error, delayMs }) => {
      console.warn(`retry ${attempt}/${maxAttempts} in ${delayMs} ms: ${error.message}`);
    },
  },
});

Which calls the client retries:

Call	Retried on
A call that changes nothing: a ConnectRPC method that the Protobuf definition marks `NO_SIDE_EFFECTS` (for example `listFunctions`, `getRun`), or a REST `GET`, `HEAD`, `PUT` or `DELETE`	Network error, timeout, `408`, `429`, `5xx`
All other calls (for example `emit`, `registerFunction`, `streams.append`)	`429` only

A retry sends the same request again. After a network error, a timeout or a 5xx, the client cannot know if the server applied a write, so it does not send the write again. A 429 is different: the server refused the request before it ran, so a retry is safe for all calls.

Behavior:

If the response has a Retry-After header and its value is longer than the backoff delay, the client waits for the Retry-After time. If there is no header, the client uses retry_after (seconds) from the error body. The wait is never longer than maxDelayMs.
The server’s retryable flag in a REST error body wins when it is present. A safe call is not retried when the flag is false (for example a 507), and error.retryable follows the flag. Without the flag, the status decides.
501 is never retried. A missing route does not appear on a retry.
A write is still sent again only on 429.
onRetry runs before each wait. onError runs one time, for the last failure.
The request timeout applies to each attempt.
To disable retry, set maxAttempts: 1.
client.kv() and client.config() do not retry.

Event Emission

emit

Emit an event to trigger workflows:

const result = await client.emit('order.placed', {
  orderId: '123',
  amount: 99.99,
});

console.log('Event ID:', result.eventId);
console.log('Run IDs:', result.runIds);

emit does not wait for the runs. It returns when the server has stored the event and created the runs. Use emitSync to wait for the runs.

Keep event names in an as const object, and use it in triggers, projections, emit, waitForEvent and streams.append. Do not repeat raw string literals. Webhook event names are set at run time, so they can use template strings.

export const OrderEvents = {
  PLACED: "order.placed",
  SHIPPED: "order.shipped",
} as const;

The function context has no client. To emit from inside a function, import createClient and call emit inside a step.run, so that the emit is memoized.

Emit options:

// Emit with a specific event version
const result = await client.emit('order.placed', data, {
  version: 2,
});

// Emit with idempotency key (deduplication)
const result = await client.emit('payment.processed', data, {
  idempotencyKey: 'payment-abc',
});

// Emit with metadata
const result = await client.emit('order.placed', data, {
  metadata: { source: 'api' },
});

triggerBatch

Emit several events in one round trip. Returns one TriggerBatchResult per event, in order. A failed item carries error (code and message, for example invalid_argument for a reserved name) and no event ID. It does not fail the call, and the other items still emit.

const results = await client.triggerBatch([
  { event: 'order.placed', data: { orderId: '1' } },
  { event: 'order.placed', data: { orderId: '2' }, idempotencyKey: 'order-2' },
]);

listEvents / getEvent / listEventNames

Read stored events. All listEvents filters are optional; paginate with nextCursor.

const page = await client.listEvents({
  names: ['order.placed'],
  since: '2026-01-01T00:00:00Z',
  limit: 50,
});
console.log(page.events, page.hasNext, page.nextCursor);

const event = await client.getEvent('evt_abc123');

// Distinct event names with counts. The server scans up to `scanCap` events
// and sets `truncated` when it stopped early.
const { names, scanned, truncated, scanCap } = await client.listEventNames();

emitSync

Emit an event and wait synchronously for every run it triggers. Uses the server-side TriggerSync endpoint, which blocks until the runs finish or the timeout elapses.

Returns EmitSyncResult[] — one element per matched trigger. An event that matches nothing returns []; that is not an error.

Run outcomes are never thrown: with N results there is no unambiguous choice of which failure to raise, and an aggregate throw would hide the runs that succeeded. Inspect status, error and waitTimedOut per element. Transport, protocol and validation errors still throw. Use invoke when you want one function and an exception on failure.

const results = await client.emitSync('order.placed', {
  orderId: '123',
  amount: 99.99,
}, { timeout: 60000, idempotencyKey: 'order-123-placed' });

for (const result of results) {
  if (result.waitTimedOut) {
    console.warn('Still running:', result.runId, result.status);
  } else if (result.status === 'failed') {
    console.error('Run failed:', result.runId, result.error?.message);
  } else {
    console.log('Output:', result.functionId, result.output);
  }
}

Options:

Parameter	Type	Description
`eventName`	`string`	Event name to emit
`data`	`unknown`	Event payload
`options.timeout`	`number`	Maximum server wait in ms (default: `30000`); the transport adds 5 seconds of response grace
`options.idempotencyKey`	`string`	Deduplication key — a repeat emit returns the original runs
`options.version`	`number`	Event schema version (#1955). Omit for `1`. Only meaningful with event-schema enforcement on

Returns: EmitSyncResult[]

Field	Type	Description
`runId`	`string`	Run ID
`functionId`	`string`	Function ID that handled the event
`status`	`RunStatus`	Final run status
`output`	`unknown`	Function return value
`error`	`{ message, code? }`	Present when the run failed
`durationMs`	`number`	Total run duration in milliseconds
`waitTimedOut`	`boolean`	The wait budget expired; the run is still going server-side

Ceiling: Node’s built-in fetch enforces a ~300 s undici headers timeout that no AbortController overrides. A wait budget above ~295 s dies at the socket. Use emit() plus a subscription for waits that long.

invoke

Run one function by function ID and wait for its result. Calls IronflowService/InvokeFunctionSync, which targets exactly that function — no trigger matching, no fan-out — and returns a single InvokeSyncResult.

Because there is exactly one run, invoke throws on the run’s outcome: RunFailedError, RunCancelledError, or non-retryable RunWaitTimeoutError (the run survives a wait timeout; poll getRun(runId)). See ADR 0067.

import { RunFailedError, RunCancelledError, RunWaitTimeoutError } from '@ironflow/core';

try {
  const result = await client.invoke('process-order', {
    data: { orderId: '123', amount: 99.99 },
    timeout: 60000,
    idempotencyKey: 'order-123',
  });

  console.log('Run ID:', result.runId);
  console.log('Status:', result.status);
  console.log('Output:', result.output);
  console.log('Duration:', result.durationMs, 'ms');
} catch (err) {
  if (err instanceof RunFailedError) {
    console.error('Run failed:', err.runId, err.message);
  } else if (err instanceof RunCancelledError) {
    console.error('Run was cancelled:', err.runId);
  } else if (err instanceof RunWaitTimeoutError) {
    console.error('Run is still active:', err.runId, err.runStatus);
  }
}

Returns: InvokeSyncResult — the EmitSyncResult fields minus waitTimedOut, which the thrown RunWaitTimeoutError reports instead.

Warning: InvokeFunctionSync ties the run’s lifetime to the request context. Aborting the call — or dropping the connection, including at Node’s ~295 s ceiling above — cancels the run server-side.

Pub/Sub

publish

Publish a message to a developer pub/sub topic. Unlike emit(), this does not trigger functions — it delivers messages directly to topic subscribers.

const result = await client.publish('notifications', {
  userId: '123',
  message: 'Your order has shipped!',
});

console.log('Event ID:', result.eventId);
console.log('Sequence:', result.sequence);

Options:

const data = { userId: 'user-123', message: 'Hello!' };
const result = await client.publish('notifications', data, {
  idempotencyKey: 'notif-abc-123',  // Optional deduplication key
});

idempotencyKey is at most 255 bytes, with no whitespace or control characters; the server rejects anything else with invalid_argument.

Returns: PublishResult

Field	Type	Description
`eventId`	`string`	Unique ID for this published message
`sequence`	`number`	JetStream sequence number

listTopics

List all active developer pub/sub topics.

const topics = await client.listTopics();
for (const t of topics) {
  console.log(`${t.name}: ${t.messageCount} messages, ${t.consumerCount} consumers`);
}

Returns: TopicInfo[]

Field	Type	Description
`name`	`string`	Topic name
`messageCount`	`number`	Number of messages in the topic
`consumerCount`	`number`	Number of active consumers
`firstMessageAt`	`Date \| undefined`	Timestamp of the first message
`lastMessageAt`	`Date \| undefined`	Timestamp of the most recent message

getTopicStats

Get detailed statistics for a specific topic, including consumer lag.

const stats = await client.getTopicStats('notifications');
console.log('Messages:', stats.messageCount);
console.log('Consumers:', stats.consumerCount);
console.log('Lag:', stats.lag);
console.log('Sequence range:', stats.firstSeq, '→', stats.lastSeq);

Returns: TopicStats

Field	Type	Description
`name`	`string`	Topic name
`messageCount`	`number`	Number of messages
`consumerCount`	`number`	Number of active consumers
`lag`	`number`	Messages pending delivery (consumer lag)
`firstSeq`	`number`	First sequence number
`lastSeq`	`number`	Last sequence number

Run Management

getRun

Get run details:

const run = await client.getRun('run_xyz789');
console.log('Status:', run.status);

getRun, listRuns, cancelRun and resumeRun return Run, whose createdAt, updatedAt, startedAt and endedAt are Date values. startedAt and endedAt are undefined until the run reaches that point. A response with a missing or unreadable required timestamp throws SchemaValidationError.

listRuns

List runs with filtering:

const result = await client.listRuns({
  functionId: 'process-order',
  status: 'completed',
  limit: 50,
  cursor: 'abc123',  // pagination
});

for (const run of result.runs) {
  console.log(`Run ${run.id}: ${run.status}`);
}
console.log('Next page:', result.nextCursor);

getRunSteps

Get the recorded steps for a run.

const { steps, count } = await client.getRunSteps('run_abc123');
for (const step of steps) {
  console.log(step.stepId, step.status, step.output);
}

steps[].createdAt, updatedAt, startedAt, endedAt, sleepUntil, waitTimeout and patchedAt are Date values. An unreadable value throws SchemaValidationError.

getRunStreams

List the entity streams a run touched.

const { entityIds } = await client.getRunStreams('run_abc123');

cancelRun

Cancel a running workflow:

const run = await client.cancelRun('run_xyz789', 'no longer needed');

resumeRun

Resume a paused or failed run:

const run = await client.resumeRun('run_xyz789');

deleteRun

Permanently delete a terminal run and its steps:

await client.deleteRun('run_xyz789');

The run must already be completed, failed, or cancelled — a run still in flight is refused with failed_precondition; cancel it first. The delete cannot be undone, and it needs the runs:delete action, which only the admin role holds.

deleteRuns

Delete terminal runs matching a filter in the current environment, and return how many THIS call removed:

const deleted = await client.deleteRuns({
  functionId: 'process-order',
  status: 'completed',
  until: new Date('2026-01-01'),
});

At least one field is required — an empty filter is rejected with invalid_argument rather than matching every run. status, when set, must be terminal, and until matches runs created before it. The server deletes in batches of 1000 and stops at 10,000 per call, so a wide filter can never become a single unbounded sweep. Re-issue the same filter until it returns 0; deleted rows no longer match, so the same call resumes where it stopped. A count below 10000 does NOT mean drained — a page comes back short whenever another transaction holds some matching rows.

redactEvent, redactStep, redactRun

Redaction replaces a stored payload with a placeholder. The row survives with its id, ordering and version intact, so replay, memoization and optimistic concurrency are unaffected; only the bytes go. It cannot be undone, and it costs events:redact — admin by default.

{"$redacted": true, "sha256": "<hex of the original bytes>", "redactedAt": "<RFC 3339>"}

All three calls are idempotent. Five things they do not reach: projection state, which is yours and is never rewritten — rebuild the projection to make it forget; a run’s copy of the event payload (input), which only RedactRun removes; a step’s own payload columns on a run redaction, and the run’s copy of a step’s output, which only the other call removes — redacting a step usually leaves a copy in the run’s output, so call both; a copy already published to a stream, which expires on its own retention window (at most 7 days), including a publish past its final read of the row when the redaction commits; and an object-storage copy awaiting successful cleanup. Run and step markers authorize repeated cleanup while the server and blob backend are available, with no cleanup deadline. Event snapshot cleanup is best effort; a failed sweep can leave copies behind without automatic retry. See Redacting an Event’s Payload.

// The event's data. Snapshots of an entity stream derived from it are dropped
// with it. No terminal-run guard.
await client.redactEvent("ev_01JD2KQ7X8Y9Z");

// The step's output, original output and error, its parent run's error,
// its attempt errors and its audit payloads. Sibling errors are unchanged.
// The argument is the step's ROW id, not its name. The run must be terminal.
await client.redactStep("step_01JD2KQ7X8Y9Z");

// The run's input, output and error, every child step's error,
// plus every attempt error and audit payload for
// the run and its steps. The run must be terminal.
await client.redactRun("run_abc123");

Error envelopes are replaced whole, including message, code, stack, details and encrypted envelopes. Run redaction leaves step outputs unchanged; use step redaction for those. Stored error placeholders survive later whole-row saves.

A non-terminal run is refused with failed_precondition: a memoized resume reads the step output and a resume reads the run input, so redacting under a live run would change results rather than surface a placeholder.

Use isRedacted in a reducer before trusting a payload. Upcasters need no guard — UpcasterRegistry skips a redacted payload and returns it untouched, so the marker reaches the reducer intact:

import { isRedacted } from "@ironflow/node";

if (isRedacted(event.data)) return state; // keep what we had

History Editing (Scoped Injection)

// Pause a running workflow at the next step boundary
const result = await client.pauseRun("run_abc123");
// result.status: "pause_requested" or "paused"

// View completed steps while paused
const state = await client.getPausedState("run_abc123");
for (const step of state.steps) {
  console.log(step.name, step.output, step.injected);
}

// Inject modified output for a step
const prev = await client.injectStepOutput(
  "run_abc123",
  "step_xyz",
  { corrected: true },
  "Manual correction"
);

// Resume the workflow
await client.resumeRun("run_abc123");

patchStep

Hot-patch a step’s output to fix a failed run:

await client.patchStep('step_abc123', { correctedValue: 42 }, 'fixed bad API response');

registerFunction

await client.registerFunction({
  id: 'my-function',
  name: 'My Function',
  triggers: [{ event: 'my.event' }],
  endpointUrl: 'http://localhost:3000/api/ironflow',
  preferredMode: 'push',
  secrets: ['STRIPE_KEY'],        // optional: secret names the function requires
  recordingProfile: 'steps',      // optional: 'all' | 'run_lifecycle' | 'steps'
  metadata: { team: 'billing' },  // optional
});

registerFunction sends only the fields you pass. It accepts these fields: id (required), name, description, triggers, retry, timeoutMs, concurrency, circuitBreaker, throttle, debounce, preferredMode, endpointUrl, actorKey, cancelOn, secrets, recording, recordingProfile and metadata.

A function registered without recording or recordingProfile records no runs, so ironflow inspect has no frames for it. A non-empty recordingProfile turns recording on. worker.start() sends these config fields for you.

Function Lifecycle

Read and manage registered functions. Every mutation bumps the function’s version and is recorded in its history.

const fn = await client.getFunction('process-order');
console.log(fn.status, fn.version, fn.triggers);

// 'active' | 'paused' | 'archived'
await client.updateFunctionStatus('process-order', 'paused');

await client.deleteFunction('process-order');

// Version history, newest first
const { entries, hasMore } = await client.listFunctionHistory('process-order', {
  limit: 20,
  fromVersion: 5,
});

const entry = await client.getFunctionAtVersion('process-order', 3);
console.log(entry.functionSnapshot);

// Restoring an earlier version is itself a new version
await client.rollbackFunction('process-order', 3, 'bad deploy');

createdAt, updatedAt and recordedAt are Date values. An unreadable value throws SchemaValidationError.

Time-Travel Debugging

Reconstruct run state at a point in time, inspect step outputs, and walk through a run’s event timeline. Useful for debugging failed or unexpected runs after the fact.

getRunStateAt

Get the reconstructed state of a run at a specific timestamp.

const state = await client.getRunStateAt('run_xyz789', new Date('2024-06-01T12:00:00Z'));

console.log('Run ID:', state.runId);
console.log('Status at timestamp:', state.status);
for (const step of state.steps) {
  console.log(`  ${step.name} (${step.stepId}): ${step.status}`, step.output);
}

Parameters:

Parameter	Type	Description
`runId`	`string`	Run ID to query
`timestamp`	`Date`	Point in time to reconstruct state at

Returns: TimeTravelRunStateSnapshot

Field	Type	Description
`runId`	`string`	Run ID
`functionId`	`string`	Function ID
`status`	`string`	Run status at the given timestamp
`input`	`unknown`	Run input, decoded from JSON
`steps`	`TimeTravelStepSnapshot[]`	Step states at the given timestamp: `stepId`, `name`, `type`, `sequence`, `status`, `output`, `error`, `originalOutput`, `startedAt`, `completedAt`, `durationMs`, `injected`, `patched`
`timestamp`	`Date`	The queried point in time
`createdAt`	`Date \| null`	When the run was created

getRunTimeline

Get the timeline of audit events for a run, ordered chronologically. Includes step transitions, status changes, and other significant events.

const events = await client.getRunTimeline('run_xyz789');
for (const e of events) {
  console.log(`[${e.timestamp}] ${e.eventType}: ${e.summary}`);
}

Returns: TimeTravelTimelineEvent[]

Field	Type	Description
`id`	`string`	Event ID
`eventType`	`string`	Type of timeline event (e.g. `step.completed`)
`stepId`	`string`	Associated step ID (empty if not step-related)
`stepName`	`string`	Associated step name
`summary`	`string`	Human-readable description
`significant`	`boolean`	Whether this event is considered a key milestone
`timestamp`	`Date`	When this event occurred

getStepOutputAt

Get the output of a specific step at a point in time. Useful for comparing step outputs before and after a patch or injection.

const output = await client.getStepOutputAt(
  'run_xyz789',
  'step_abc123',
  new Date('2024-06-01T12:05:00Z')
);

console.log('Step ID:', output.stepId);
console.log('Status at time:', output.status);
console.log('Output at time:', output.output);

Parameters:

Parameter	Type	Description
`runId`	`string`	Run ID
`stepId`	`string`	The step name the code used (`charge`), or the full step ID from `getRunSteps` (`{runId}:charge:0`)
`timestamp`	`Date`	Point in time to query

Returns: TimeTravelStepOutputSnapshot. Throws a not-found error when the step did not exist at that time.

Field	Type	Description
`stepId`	`string`	Step ID
`status`	`string`	Step status at the given timestamp
`output`	`unknown`	Step output at the given timestamp, decoded from JSON
`originalOutput`	`unknown \| null`	Output before a patch, or `null`
`patched`	`boolean`	The output was patched
`injected`	`boolean`	The output was injected

Audit Trail

listAuditEvents

Query the environment-wide audit stream:

const page = await client.listAuditEvents({ eventType: 'run.failed', limit: 50 });

getAuditTrail

Get the audit trail for a run — a list of recorded events capturing what happened at each stage of execution.

const trail = await client.getAuditTrail('run_xyz789');
for (const event of trail.events) {
  console.log(`[${event.createdAt}] ${event.eventType}`, event.payload);
}

Pass options to filter and page: eventType, fromTimestamp, toTimestamp, limit and cursor (the previous page’s nextCursor):

const page = await client.getAuditTrail('run_xyz789', { limit: 50, cursor: trail.nextCursor });

Returns: AuditTrailResult, the same shape as listAuditEvents: events, totalCount and nextCursor.

Event field	Type	Description
`id`	`string`	Event ID
`runId`	`string`	Run ID
`functionId`	`string`	Function ID
`stepId`	`string \| undefined`	Step ID, for step events
`eventType`	`string`	Audit event type (e.g. `run.created`, `step.completed`)
`payload`	`Record<string, unknown>`	Event-specific payload
`createdAt`	`Date`	When the event was recorded

createdAt is a Date. An unreadable value throws SchemaValidationError. fromTimestamp and toTimestamp take a Date or a string. Page with cursor, not with a returned createdAt. Rows hold finer than milliseconds, so toTimestamp: event.createdAt excludes that row.

Serve Handler

serve

Create an HTTP handler for push mode.

import { serve } from '@ironflow/node';

const handler = serve({
  // Required: functions to serve
  functions: [fn1, fn2, fn3],

  // Optional: signing key for request verification
  signingKey: process.env.IRONFLOW_SIGNING_KEY,

  // Optional: skip signature verification (dev only)
  skipVerification: false,

  // Optional: target environment (an environment ID; default: IRONFLOW_ENV, else no header)
  environment: 'env_default_staging',

  // Optional: event definitions for automatic upcasting
  eventDefinitions: registry,

  // Optional: transport diagnostic logger, or false to disable
  logger: console,

  // Optional: application execution logger, or false to silence it
  userLogger: false,

  // Optional: Ironflow server URL for emitting webhook events
  serverUrl: 'http://localhost:9123',

  // Optional: webhook sources to handle
  webhooks: [stripeWebhook],
});

serve() returns a handler of type (request: Request | NodeRequest, responseOrContext?) => Promise<Response | void>. With a Web Request it returns a Response. With Node req and res objects it writes to res and returns nothing.

Framework Integration

Next.js App Router:

import { serve } from '@ironflow/node';

export const POST = serve({ functions: [myFunction] });

Express:

import express from 'express';
import { createHandler } from '@ironflow/node';

const app = express();

// createHandler is an alias for serve()
app.post('/api/ironflow', createHandler({ functions: [myFunction] }));

Worker

createWorker

Create a pull mode worker.

import { createWorker } from '@ironflow/node';

const worker = createWorker({
  // Optional: server URL (default: IRONFLOW_SERVER_URL or http://localhost:9123)
  serverUrl: 'http://localhost:9123',

  // Required: functions to execute
  functions: [fn1, fn2],

  // Optional: API key for authentication (default: IRONFLOW_API_KEY env var)
  apiKey: process.env.IRONFLOW_API_KEY,

  // Optional: target environment (an environment ID; default: IRONFLOW_ENV, else no header)
  environment: 'env_default_staging',

  // Optional: event definitions for automatic upcasting
  eventDefinitions: registry,

  // Optional: max concurrent jobs (default: 10)
  maxConcurrentJobs: 10,

  // Optional: worker labels for routing
  labels: { region: 'us-east-1' },

  // Optional: heartbeat interval in ms (default: 30000)
  heartbeatInterval: 30000,

  // Optional: reconnect delay in ms (default: 5000)
  reconnectDelay: 5000,

  // Optional: step-checkpoint debounce in ms (default: 1000, 0 disables)
  checkpointInterval: 1000,

  // Optional: seconds the server may hold an idle job poll (default: 20, max 25, 0 disables)
  pollWaitSeconds: 20,

  // Optional: drain deadline in ms (default: 30000)
  drainTimeout: 30000,

  // Optional: transport diagnostic logger, or false to disable
  logger: console,

  // Optional: application execution logger, or false to silence it
  userLogger: false,
});

Authentication failures stop the worker. A 401 or 403 from function registration, worker registration, or job polling rejects start() with UnauthenticatedError / UnauthorizedError after logging the API-key and bootstrap-key-file guidance. The worker does not retry on the reconnectDelay cadence — a missing or revoked key does not fix itself, so a supervisor restart (with the key set) is the recovery path. Every other connection error still reconnects as before.

Step checkpointing. While a job runs, the worker debounces completed step results and posts them to the server. If the worker is killed, the reclaimed run memoizes the checkpointed steps instead of re-executing them — at most one checkpointInterval window of progress is lost. Set checkpointInterval: 0 to disable it and report steps only in the final update.

Checkpointing is best-effort and never fails a job: batches are capped at 500 steps, a failed flush retries with exponential backoff, and the steps stay buffered so they still go out in the final update. If the server rejects a checkpoint with any 4xx — including an older server that does not implement checkpointing — the worker logs once, stops checkpointing that job, and runs it to completion normally. Polling workers only; createStreamingWorker ignores the option. The streaming worker does not need it: it sends each step.run result on the stream when the step completes, so the engine already has the completed steps if the worker is killed.

Workers track active assignments by job ID and execution sequence. If recovery assigns the same job to this worker with a new sequence, both executions count toward capacity and drain during normal operation. An older completion cannot remove the newer execution from heartbeats. An exact redelivery of an active sequence is ignored. This applies to polling and ConnectRPC streaming workers; streaming cancellation by job ID signals each locally tracked execution of that job.

When a streaming connection drops, the Node worker aborts its execution signals and immediately removes those executions from tracking. A handler waiting on work outside SDK steps may still be running. Those handlers no longer count toward capacity or drain after reconnect. Go and Python streaming workers retain their execution slots across ordinary connection loss. A forced stop or drain deadline also bounds how long a worker waits for handlers.

Worker Methods

// Start the worker. The promise settles only after the worker stops, so
// call drain() and stop() from other code, such as a timer or a signal handler.
await worker.start();

// Graceful drain (wait up to drainTimeout for active jobs, then cancel them)
await worker.drain();

// Force stop
worker.stop();

Shutdown. drain() accepts no new jobs and waits up to drainTimeout (default 30 seconds) for the active jobs. At the deadline it cancels the jobs that remain and stops, so the engine can reclaim their leases. start() connects SIGINT and SIGTERM to drain(). One listener serves all the workers in the process, from createWorker and from createStreamingWorker. When all of them have drained, the process exits. If your app has a listener of its own for the signal, added with process.on or process.once, the process does not exit and your app owns the shutdown.

createStreamingWorker

Create a worker using gRPC bidirectional streaming (lower latency).

Note: Import from the separate entry point to avoid loading protobuf dependencies.

import { createStreamingWorker } from '@ironflow/node/worker-streaming';

const worker = createStreamingWorker({
  serverUrl: 'http://localhost:9123',
  functions: [myFunction],
});

await worker.start();

Upgrade order. Upgrade the engine before the streaming workers, and upgrade Go and Node streaming workers in the same rollout. An older Go or Node streaming SDK against a newer engine reads a failed invoke as success with empty output.

Differences from createWorker.

No step checkpoints. checkpointInterval is ignored. Each step.run result goes on the stream when the step completes.
A job the worker cannot run. The engine tracks maxConcurrentJobs, but it can send more jobs than that, for example after a reconnect. The worker refuses a job that arrives when it is full or during a drain. It sends a nack, and the engine makes the job runnable again at once. A refusal does not use a retry attempt. The engine does not send more jobs to a draining worker. For each refusal from a full worker, the engine holds back one of that worker’s slots. It frees the slot when the worker’s heartbeat shows a free slot. With an engine older than the nack, the engine ignores the refusal. It then recovers the job after the lease expires (about 90 s).
Graceful close. When the active jobs are done, a drain sends the queued results, half-closes the stream and waits for the engine to close it, within the same deadline. stop() closes the stream immediately.
Stream lost during a drain. The worker stops and does not reconnect. Without a stream it cannot report results.
Drain timeout. An engine shutdown message starts a drain. The worker uses the drain timeout in the message (the engine sends 25 seconds). If the message has no drain timeout, the worker uses drainTimeout. If a drain already runs, its deadline does not change. The Python SDK default for drain_timeout is 60 seconds, not 30.

Worker with Projections

Workers can run projection runners alongside functions. The worker automatically starts a ProjectionRunner for each projection, using streaming mode with automatic fallback to polling. Both createWorker and createStreamingWorker do this — projections are plain HTTP and independent of how the worker receives its jobs.

import { createWorker, createProjection } from '@ironflow/node';

const orderStats = createProjection({
  name: 'order-stats',
  events: ['order.created', 'order.completed'],
  initialState: () => ({ total: 0, completed: 0 }),
  handler: (state, event) => {
    if (event.name === 'order.created') return { ...state, total: state.total + 1 };
    if (event.name === 'order.completed') return { ...state, completed: state.completed + 1 };
    return state;
  },
});

const worker = createWorker({
  functions: [processOrder],
  projections: [orderStats],
});

await worker.start();
// Projection runners start automatically with streaming-first fallback

The projections option has the type IronflowProjection[], which means IronflowProjection<unknown, unknown>. The handler of a managed projection takes a typed state, so TypeScript does not accept IronflowProjection<MyState, MyEvent> for that element. Cast each projection, for example projections: [orderStats as IronflowProjection], or the array, for example [a, b] as IronflowProjection[]. Import IronflowProjection from @ironflow/node.

Projections

Projection-only workers

Pass functions: [] and projections: [projection] to createWorker() or createStreamingWorker() to run projections without a dummy function. Projection runners use their own HTTP streams and leases. These workers do not register a function worker, open a job stream, poll jobs, or send function-job heartbeats, so they do not appear as function workers in the worker list. stop() and drain() stop the runners; start() resolves after pending projection saves finish.

createProjection

Define a projection for use with a worker.

import { createProjection } from '@ironflow/node';

// Managed projection (pure reducer, Ironflow stores state)
const stats = createProjection({
  name: 'order-stats',
  events: ['order.created', 'order.completed'],
  initialState: () => ({ total: 0 }),
  handler: (state, event, ctx) => {
    return { ...state, total: state.total + 1 };
  },
  // Optional:
  batchSize: 100,          // Events per batch (default: 100)
  maxRetries: 3,           // Retries for a failing event (default: 3)
  partitionKey: '$.customerId', // JSONPath for partitioning
});

// External projection (side effects, you manage storage)
const searchSync = createProjection({
  name: 'search-sync',
  events: ['product.*'],
  handler: async (event, ctx) => {
    await searchIndex.upsert(event.data);
  },
  // No initialState → external mode
});

createProjectionRunner

For advanced use cases, create a standalone projection runner outside of a worker:

import { createProjectionRunner } from '@ironflow/node';

const runner = createProjectionRunner({
  projection: myProjection,
  baseUrl: 'http://localhost:9123',
  headers: { Authorization: `Bearer ${process.env.IRONFLOW_API_KEY}` },
  logger: console,
  signal: abortController.signal,
});

// Streams, and polls when the server does not support streaming.
await runner.run();

The runner sends headers, plus Content-Type: application/json, on every request. The server reads the key from Authorization: Bearer <key>, and the environment from X-Ironflow-Environment-ID. createWorker builds the same headers for the runners it starts.

run() holds one projection lease (ADR 0120) for both modes. startStreaming() only streams, and start() only polls. When you call startStreaming() and it throws StreamingUnsupportedError, the runner releases its lease. A start() call after it gets the lease again. The lease is required: on a server older than v0.44.0, which has no lease RPCs, run() rejects with ServerTooOldError (code SERVER_TOO_OLD; match the code, not instanceof) and sends no poll, stream, save or ack.

StreamingUnsupportedError

Thrown by startStreaming() when the server does not support the StreamProjectionEvents RPC. Catch this error to fall back to polling mode, or call run(), which falls back by itself.

The runner reads the Connect error code. The code can come from the HTTP status or from the end-of-stream trailer.

Connect code	HTTP status	Result
`unimplemented`, `not_found`	404	`StreamingUnsupportedError`. Fall back to polling.
`unauthenticated`	401	`UnauthenticatedError`. No fallback.
`permission_denied`	403	`UnauthorizedError`. No fallback.
`failed_precondition`	400	Delivery refused, such as a projection bound to a PUBSUB stream that was recreated (ADR 0164). The runner stops.
Any other code, such as `internal`	Any other status, such as 501	Stream failure. The runner reconnects with backoff from 2 s to 30 s.

Auth failures and a refused delivery reject run(), startStreaming() and start(), from a poll as from a stream. A retry cannot fix them.

import { StreamingUnsupportedError } from '@ironflow/node';

ProjectionRunnerConfig

interface ProjectionRunnerConfig {
  projection: IronflowProjection;  // Projection definition
  baseUrl: string;                  // Ironflow server URL
  headers: Record<string, string>; // Request headers (auth, etc.)
  logger: Logger;                   // Logger instance
  signal?: AbortSignal;            // For graceful shutdown
  after?: Promise<void>;           // Intake waits for it: the runners this one replaces have stopped
}

See the Projections explanation for full documentation on managed vs external projections, partitioning, and real-time subscriptions.

KV Store

client.kv()

Get a KV client for bucket management and key operations:

const kv = client.kv();

createBucket

const info = await kv.createBucket({
  name: "sessions",
  description: "User session data",
  ttlSeconds: 3600,
  maxValueSize: 65536,
  history: 3,
});

deleteBucket

await kv.deleteBucket("sessions");

listBuckets

const buckets = await kv.listBuckets();
for (const b of buckets) {
  console.log(`${b.name}: ${b.values} keys, ${b.bytes} bytes`);
}

getBucketInfo

const info = await kv.getBucketInfo("sessions");
// info.name, info.values, info.bytes, info.history, info.created_at

created_at is a Date. An unreadable value throws SchemaValidationError.

Bucket Handle

Get a bucket handle for key operations:

const bucket = kv.bucket("sessions");

get

Returns an entry whose value contains the stored bytes encoded as base64.

const entry = await bucket.get("user.123");
// entry.key, entry.value, entry.revision, entry.created_at, entry.operation

created_at is a Date, the time this revision was written. A deleted or purged key returns 404, so an entry always has one. Compare-and-swap uses revision, not this time.

getJSON

getJSON(key: string): Promise<KVEntry> decodes the stored bytes as UTF-8 JSON and preserves the entry metadata.

const entry = await bucket.getJSON("user.123");
console.log(entry.value); // The stored JSON value, typed as unknown

Validate the application-specific shape before using value. Invalid base64, UTF-8, or JSON throws IronflowError with code INVALID_KV_JSON. HTTP errors retain their existing codes. Plain string writes are raw text and are not automatically JSON-quoted.

put

Unconditional write. Returns the new revision:

const { revision } = await bucket.put("user.123", { name: "Alice" });

create (If-Not-Exists)

Write only if the key does not exist. Throws on conflict (HTTP 412):

const { revision } = await bucket.create("user.123", { name: "Alice" });

update (Compare-and-Swap)

Write only if the revision matches. Throws on mismatch (HTTP 412):

const { revision } = await bucket.update("user.123", newValue, entry.revision);

delete

Soft-delete (tombstone):

await bucket.delete("user.123");

purge

Permanently remove key and all history:

await bucket.purge("user.123");

listKeys

List keys with optional wildcard filter:

const keys = await bucket.listKeys("user.*");
// Pass no argument for all keys

watch

Watch a bucket for real-time key changes via WebSocket. On connect, onUpdate receives the current entry of every key the watch covers. After that it fires on each put and delete. The optional key option narrows the watch to one key or a pattern such as "user.*". The REST reference gives the stream format and errors. Requires Node.js 22+ (uses the built-in global WebSocket).

const watcher = bucket.watch({
  onUpdate: (event) => {
    // event.value is base64; it is "" for a delete
    console.log(`${event.key} ${event.operation}: rev ${event.revision}`);
  },
  onError: (err) => console.error(err.message),
  onClose: () => console.log("Watch ended"),
}, { key: "user.*" });

// Stop watching
watcher.stop();

See the KV Store Guide for detailed usage patterns.

File Storage

client.files()

Get a files client for buckets, files, signed URLs and events. See the Files guide.

const files = client.files();                        // files() is a METHOD, not a property
await files.createBucket("inbox", { emitEvents: true, allowSignedUrls: true });

const inbox = files.bucket("inbox");
const info = await inbox.put("scans/a.pdf", bytes, { contentType: "application/pdf" });
const file = await inbox.get("scans/a.pdf", { ifMatch: info.etag });
const text = await file.text();                      // or arrayBuffer(), or read file.body

FilesClient methods:

Method	Description
`createBucket(name, config?)`	Create a bucket. `config`: `maxObjectBytes`, `allowedContentTypes`, `emitEvents`, `allowSignedUrls`.
`listBuckets()`	`FileBucketInfo[]`.
`getBucket(name)`	One bucket’s configuration.
`updateBucket(name, config)`	Change the set fields. The change applies to new writes only.
`deleteBucket(name)`	Delete an empty bucket. A bucket with files throws `ConflictError`.
`bucket(name)`	A `FileBucketHandle`.

FileBucketHandle methods:

Method	Description
`put(path, body, { contentType, contentLength?, metadata?, ifMatch?, ifNoneMatch? })`	Upload. Returns `FileInfo`.
`get(path, { ifMatch?, range? })`	`{ body, contentType, etag, size, arrayBuffer(), text() }`.
`info(path)`	A file’s metadata, without the bytes.
`list({ prefix?, delimiter?, cursor?, limit? })`	`{ files, prefixes, nextCursor }`.
`delete(path)`	Delete a file. A missing path is not an error.
`move(from, to, { toBucket?, ifMatch? })`	Rename one file.
`copy(from, to, { toBucket? })`	Copy one file.
`movePrefix(from, to)`	Move every file under a prefix (both end in `/`). Returns the count.
`signUpload(path, { ttlSeconds?, maxBytes?, contentType?, createOnly? })`	`{ url, expiresAt }` for a credential-free upload. `createOnly` makes the PUT fail with `412` if the path exists.
`signDownload(path, ttlSeconds?)`	`{ url, expiresAt }` for a credential-free download.

FileInfo and FileBucketInfo createdAt and updatedAt, and SignedUrl.expiresAt, are Date values. An unreadable value throws SchemaValidationError.

body is a Uint8Array, ArrayBuffer, Blob or string. It can also be a Node Readable or a ReadableStream. A stream needs contentLength. A stream upload is sent once: it is not retried after a 503. The other bodies are retried. put has no fixed timeout.

Errors: PreconditionFailedError (412), PayloadTooLargeError (413), UnsupportedMediaTypeError (415) and ConflictError (409). A 404 is an IronflowError with code not_found. Import the three file error classes from @ironflow/node.

Config Management

client.config()

Get a config client for environment-scoped configuration:

const config = client.config();

set

Full replacement of a config document:

const { revision } = await config.set("app-settings", {
  theme: "dark",
  maxRetries: 3,
});

create and update

Conditional writes. create fails if the name exists. update replaces the config only if revision is its current revision. Both throw an IronflowError on a failed condition (HTTP 412).

const { revision } = await config.create("app-settings", { theme: "dark" });
await config.update("app-settings", { theme: "light" }, revision);

patchIf(name, data, revision) and deleteIf(name, revision) apply the same condition to patch and delete.

get

const entry = await config.get("app-settings");
// entry.name, entry.data, entry.revision, entry.updatedAt

updatedAt is a Date, to the second on get and list. An unreadable value throws SchemaValidationError. Compare-and-swap uses revision, not this time.

patch

Shallow merge — only specified keys are updated:

const { revision } = await config.patch("app-settings", {
  maxRetries: 5,
  timeout: 30,
});

list

const configs = await config.list();
for (const entry of configs) {
  console.log(`${entry.name}: rev ${entry.revision}`);
}

delete

Delete a config. Idempotent — succeeds silently if the config does not exist.

await config.delete("app-settings");

watch

Watch a config document for real-time changes via WebSocket. On connect, onUpdate receives the current document if it exists. After that it fires each time the document is created, updated or deleted. Returns a ConfigWatcher handle with a stop() method to close the connection. The REST reference gives the stream format and errors.

Requires Node.js 22+ (uses the built-in global WebSocket).

const watcher = config.watch("app-settings", {
  onUpdate: (event) => {
    console.log("Config updated:", event.name);
    console.log("New data:", event.data);
    console.log("Revision:", event.revision);
    console.log("Updated at:", event.updatedAt);
  },
  onError: (err) => {
    console.error("Watch error:", err.message);
  },
  onClose: () => {
    console.log("Watch connection closed");
  },
});

// Stop watching when done
watcher.stop();

Parameters:

Parameter	Type	Description
`name`	`string`	Config document name to watch
`callbacks.onUpdate`	`(event: ConfigWatchEvent) => void`	Called with the current document on connect, then on each create, update or delete
`callbacks.onError`	`(error: Error) => void`	Optional. Called when an error occurs
`callbacks.onClose`	`() => void`	Optional. Called when the WebSocket connection closes

ConfigWatchEvent fields:

Field	Type	Description
`type`	`string`	Always `"config_update"` for data events
`name`	`string`	Config document name
`data`	`Record<string, unknown> \| null`	Current config data; `null` after a delete
`revision`	`number`	KV revision number after this update
`updatedAt`	`Date`	When the update happened. A frame with an unreadable value goes to `onError`, and the watch stays open

Returns: ConfigWatcher — call watcher.stop() to close the connection.

See the Config Management explanation for detailed usage patterns.

Secrets

Secrets are encrypted, environment-scoped values (API keys, passwords, tokens) that can only be read at runtime inside functions. They are managed via the CLI or REST API and accessed through the SDK’s read-only interface.

Declaring Secrets

Declare which secrets a function needs in the function config:

import { ironflow } from '@ironflow/node';

const myFunction = ironflow.createFunction(
  {
    id: 'my-function',
    triggers: [{ event: 'my.event' }],
    secrets: ['API_KEY', 'DB_PASSWORD'],
  },
  async ({ event, step, secrets }) => {
    // Access resolved secret values
    const apiKey = secrets.get('API_KEY');
    // Use apiKey...
  }
);

The engine resolves only the declared secrets at execution time. Undeclared secrets are not available.

SecretsClient

The secrets parameter in the function handler implements the SecretsClient interface:

Method	Signature	Description
`get`	`get(name: string): string`	Get a secret’s value. Throws if not found.
`has`	`has(name: string): boolean`	Check if a secret was resolved.

async ({ event, step, secrets }) => {
  // Check if an optional secret exists
  if (secrets.has('OPTIONAL_KEY')) {
    const key = secrets.get('OPTIONAL_KEY');
    // Use key...
  }

  // Get a required secret (throws if not set)
  const password = secrets.get('DB_PASSWORD');
}

createSecretsClient(secrets?: Record<string, string>) builds a SecretsClient over a plain map. The workers and serve() use it for you. Call it yourself to unit-test a helper that takes a SecretsClient.

See the Secrets Management explanation for detailed usage patterns.

Subscriptions

Real-time event subscriptions over WebSocket. Subscribe to system events (runs, functions, secrets) or user events with pattern matching.

createSubscriptionClient

Create a subscription client instance.

import { createSubscriptionClient } from "@ironflow/node";

const client = createSubscriptionClient({
  serverUrl: "http://localhost:9123",
  // Optional:
  apiKey: "my-api-key",         // API key when auth is enabled
  environment: "env_default_production",     // Environment ID
  autoReconnect: true,           // Auto-reconnect on disconnect (default: true)
  reconnectDelay: 1000,          // Initial reconnect delay in ms (default: 1000)
  maxReconnectDelay: 30000,      // Max reconnect delay in ms (default: 30000)
  reconnectBackoff: 1.5,         // Reconnect backoff multiplier (default: 1.5)
  connectionTimeout: 10000,      // Connection timeout in ms (default: 10000)
});

createRunVersionFilter

Frames can arrive out of order across nodes. createRunVersionFilter() drops strictly lower positive versions, ranking each run and each step independently by its full identity, including dots in step IDs. Ties and zero or absent versions pass without lowering the remembered version. A deletion is kept and remembers the run ID; subsequent run and step frames for that ID are dropped, including unversioned frames. Other topics pass. One filter belongs to one subscription. Its shared cache holds at most maxRuns run and step identities (default 10,000); eviction forgets both ordering and deletion history.

import { createRunVersionFilter, type RunEvent } from "@ironflow/node";

const keep = createRunVersionFilter();
await client.subscribe("system.run.>", {
  onEvent: (event) => {
    if (keep(event)) render(event.data as RunEvent);
  },
});

RunEvent is the typed payload of a run frame, including version.

connect / close

await client.connect();
// ... use subscriptions ...
client.close();

Subscribe to events matching a pattern. Returns a Subscription handle.

const sub = await client.subscribe("system.run.>", {
  onEvent: (event) => {
    console.log(event.topic, event.data);
  },
  onError: (err) => {
    console.error(err.code, err.message);
  },
  // Optional:
  replay: 10,             // Replay last N events
  includeMetadata: true,  // Include event metadata
  filter: "status:failed", // Server-side filter expression
});

// Unsubscribe when done
sub.unsubscribe();

replay applies only to the initial subscribe request. Reconnects preserve the filter, consumer group, metadata, acknowledgment, backpressure, and namespace options but omit the original replay count. A fan-out subscription without a cursor reconnects at the current tail and can miss events published while it was offline. Use startAfterSequence or a consumer group when that gap matters.

onEvent is not awaited. An async handler is fine — manual acknowledgment needs one — but it runs unsupervised: the next event can be delivered before it settles, so there is no ordering or backpressure guarantee, and a rejection becomes an unhandled rejection rather than reaching onError. Catch inside the handler, including around ack and nak, and serialize the work yourself if you need ordering or a persisted cursor.

subscribeEntityStream

Subscribe to the live event feed of one entity stream. It builds the entity:{entityType}.{entityId}.> pattern and passes each event to onEvent as a StreamEvent.

const sub = await client.subscribeEntityStream("order-123", {
  entityType: "order",   // required
  onEvent: (event) => console.log(event.name, event.entityVersion),
  onError: (err) => console.error(err.message),
  replay: 100,           // optional: replay last N events
});

sub.unsubscribe();

Ackable Subscriptions

For consumer groups with manual acknowledgment:

const sub = await client.subscribe("order.*", {
  ackMode: "manual",
  consumerGroup: "processors",
  onEvent: async (event) => {
    try {
      await processOrder(event.data);
      await sub.ack(event.eventId!);
    } catch (err) {
      await sub.nak(event.eventId!, 5000); // Redeliver after 5s
    }
  },
});

client.joinConsumerGroup(groupName, pattern, options) is shorthand for subscribe with consumerGroup set to groupName and ackMode: "manual". It returns an ackable subscription.

Consumer group stream

@ironflow/node/consumer-group-stream joins a consumer group over one PubSubService.SubscribeBidirectional stream. WebSocket stays the default; this subpath is opt-in. The group must exist.

import { joinConsumerGroupStream } from "@ironflow/node/consumer-group-stream";

const sub = await joinConsumerGroupStream(
  { serverUrl: "http://localhost:9123", apiKey: process.env.IRONFLOW_API_KEY },
  "processors",
  {
    onEvent: async (event) => {
      await processOrder(event.data);
      await sub.ack(event.eventId!);
    },
  },
);

joinConsumerGroupStream(config, groupName, callbacks) resolves after the server confirms the join. It rejects with the stream error, for example NotFound for a missing group. The SDK holds events until the call returns, so sub is set when onEvent runs.

The stream needs a server that includes this change (released after v0.47.0). Against an older server the join fails with a ServerTooOldError (code SERVER_TOO_OLD) that names the first server version with the stream (0.48.0). The SDK does not retry it. Use the WebSocket transport there.

`config` field	Default	Description
`serverUrl`	required	Server URL. `http://` uses h2c.
`apiKey`	`IRONFLOW_API_KEY`	API key.
`environment`		Environment.
`namespace`	`"default"`	Consumer group namespace.
`logger`		Warnings go here. `false` silences them.
`reconnectDelay`	`1000`	First reconnect delay, in ms.
`maxReconnectDelay`	`30000`	Largest reconnect delay, in ms.
`reconnectBackoff`	`1.5`	Backoff multiplier.
`ackTimeoutMs`	`30000`	How long `ack`, `nak` and `term` wait for the server result.
`idleTimeoutMs`	`90000`	Longest wait for any frame. A longer silence counts as a dead connection, and the stream reconnects.

onEvent may return a Promise, and the SDK awaits it, so events are handled one at a time, in order. A handler that rejects goes to onError, and the SDK does not ack the event. In an AUTO group a slow handler slows the server through HTTP/2 flow control.

ack, nak and term resolve when the server applies them. They reject with:

AckError: the server refused the ack (invalid_argument, not_found or internal). An internal ack is safe to retry; a not_found ack (the group or member is gone) is not. A nak delay outside 0 to 2³¹−1 ms rejects with invalid_argument without a send; a fractional delay is rounded.
StaleAckError: the event arrived before a reconnect, the stream ended while the ack was pending, or this stream has already settled every copy of the event. In the first two cases the server redelivers the event.
SubscriptionClosedError: after unsubscribe() or a terminal error.

In an AUTO group they do nothing. A retryable disconnect reconnects with backoff. Unauthenticated, PermissionDenied, NotFound and InvalidArgument end the subscription and go to onError.

patterns

Pattern helpers for common subscription topics:

import { patterns } from "@ironflow/node";

patterns.allRuns()              // "system.run.>"
patterns.run("run_123")         // "system.run.run_123.>"
patterns.runLifecycle("run_123") // "system.run.run_123.*"
patterns.runSteps("run_123")    // "system.run.run_123.step.>"
patterns.allFunctions()         // "system.function.>"
patterns.function("my-fn")      // "system.function.my-fn.>"
patterns.allSecrets()           // "system.secret.*"
patterns.secret("API_KEY")      // "system.secret.API_KEY.*"
patterns.secretAction("updated") // "system.secret.*.updated"
patterns.allUserEvents()        // "events:>"
patterns.userEvent("order.placed") // "events:order.placed"
patterns.topic("notifications") // "topic:notifications"
patterns.allTopics()            // "topic:>"

Connection State

client.onConnectionChange((state) => {
  // state: "disconnected" | "connecting" | "connected" | "reconnecting"
  console.log("Connection:", state);
});

client.onError((err) => {
  console.error("Global error:", err.code, err.message);
});

console.log(client.isConnected);     // boolean
console.log(client.connectionState); // ConnectionState

Entity Streams

streams.append

Appends an event to an entity stream.

const result = await client.streams.append("order-123", {
  name: "order.item_added",
  data: { itemId: "item-789" },
  entityType: "order",
}, { expectedVersion: 4 });

// result.entityVersion = 5
// result.eventId = "evt-..."

append takes one event per call, not an array. It returns { entityVersion, eventId } and no sequence number. The third argument takes expectedVersion (optimistic concurrency), idempotencyKey, version (the event schema version, default 1) and metadata.

An entity ID must use only letters, digits and the characters -, _, ., : and ~. The server rejects /, ?, #, %, & and white space when you write. To scope a stream to a tenant, put the tenant in one segment, for example ${tenantId}-issue-${issueId}. Or filter by entityType.

streams.read

Reads events from an entity stream.

const { events, totalCount } = await client.streams.read("order-123", {
  fromVersion: 1,
  limit: 50,
  direction: "forward",
});

streams.getInfo

Returns stream metadata, or null if no events have been written to this stream yet.

const info = await client.streams.getInfo("order-123");
// info.entityId, info.entityType, info.version, info.eventCount
// info is null for streams with no events — safe to pass expectedVersion: 0 to append()

Returns: Promise<StreamInfo | null>

streams.listStreams / streams.getEntityHistory

const streams = await client.streams.listStreams();
const history = await client.streams.getEntityHistory('order-123', { to: '2026-10-06T23:59:59.999999999Z' });

timestamp is a Date. Pass timestampExact, not timestamp, as a from or to bound: stored timestamps are finer than a Date can hold, so a bound taken from timestamp can drop the entry it came from.

streams.createSnapshot

Create a snapshot of the materialized state at a specific stream version. Use snapshots to speed up state reconstruction for long-lived entity streams by avoiding a full event replay.

const { snapshotId } = await client.streams.createSnapshot("order-123", {
  entityType: "order",
  entityVersion: 42,
  state: { status: "shipped", items: [...] },
});

console.log("Snapshot ID:", snapshotId);

Parameters:

Parameter	Type	Description
`entityId`	`string`	Entity ID
`input.entityType`	`string`	Entity type (e.g. `"order"`). Must match the stream’s own type — a mismatch is rejected
`input.entityVersion`	`number`	Stream version this snapshot represents
`input.state`	`Record<string, unknown>`	Materialized state at this version

Returns: { snapshotId: string }

streams.getSnapshot

Get the latest snapshot at or before a given version. Returns the closest snapshot without exceeding the requested version, so you can replay only the delta of events from the snapshot forward.

const snapshot = await client.streams.getSnapshot("order-123");
// Or get snapshot at or before a specific version:
const snapshotBeforeVersion = await client.streams.getSnapshot("order-123", { beforeVersion: 40 });

console.log("Snapshot version:", snapshot.entityVersion);
console.log("State:", snapshot.state);
console.log("Created at:", snapshot.createdAt);

Parameters:

Parameter	Type	Description
`entityId`	`string`	Entity ID
`options.beforeVersion`	`number`	Return snapshot at or before this version (default: latest)

Returns: StreamSnapshot

Field	Type	Description
`snapshotId`	`string`	Snapshot ID
`entityId`	`string`	Entity ID
`entityType`	`string`	Entity type
`entityVersion`	`number`	Stream version this snapshot represents
`state`	`Record<string, unknown>`	Materialized state
`createdAt`	`Date`	When the snapshot was created

streams.delete

Delete an entity stream. Appends a $stream.deleted tombstone at the stream’s current version + 1 and drops its snapshots; later appends are refused with ConnectRPC code failed_precondition. With purge, every event row below the tombstone is deleted too — the tombstone row itself always stays. Irreversible. Requires streams:delete (admin by default).

const { entityVersion } = await client.streams.delete("order-123");
console.log("Tombstone at version:", entityVersion);

// Also delete the events below the tombstone
await client.streams.delete("order-123", { purge: true });

Parameters:

Parameter	Type	Description
`entityId`	`string`	Entity ID
`options.purge`	`boolean`	Also delete every event below the tombstone (default `false`)

Returns: { entityVersion: number } — the tombstone’s version, which is the stream’s final version.

Deleting an already-deleted stream throws failed_precondition when purge is false. With purge true it re-runs the purge and returns the existing tombstone’s version, so a delete whose cleanup failed can be retried.

Projections do not see the tombstone unless they registered $stream.deleted in their event list — delivery is filtered by event name, and nothing rewrites projection state. See Deleting a Stream.

streams.listStreams

List all entity streams registered in the system.

const streams = await client.streams.listStreams();
for (const s of streams) {
  console.log(`${s.entityType}/${s.entityId}: v${s.version}, ${s.eventCount} events`);
}

Returns: StreamListEntry[]

Field	Type	Description
`entityId`	`string`	Entity ID
`entityType`	`string`	Entity type
`version`	`number`	Current stream version
`eventCount`	`number`	Total number of events
`lastEventAt`	`Date`	When the last event was written

streams.getEntityHistory

Get the full event history for an entity as a flat array, oldest first. The SDK follows the server’s page cursor, so the result is complete for any number of events. Use streams.read() to read one page or to read newest first.

const history = await client.streams.getEntityHistory("order-123");
for (const entry of history) {
  console.log(`v${entry.version} [${entry.timestamp}] ${entry.eventName}`, entry.data);
}

// The events that make up the entity's state at a point in time
const asOfTuesday = await client.streams.getEntityHistory("order-123", {
  to: "2026-10-06T23:59:59.999999999Z", // end of Tuesday, UTC
});

Options: EntityHistoryOptions

Field	Type	Description
`fromVersion`	`number`	Start at this entity version (inclusive)
`from`	`Date \| string`	Only events at or after this time
`to`	`Date \| string`	Only events at or before this time
`limit`	`number`	Return at most this many entries. Unset reads the whole history

A string bound must be RFC 3339 and is sent unchanged, so it keeps nanosecond precision. A Date has milliseconds only, so to: new Date(entry.timestamp) can leave out that entry; pass entry.timestamp itself.

Returns: EntityHistoryEntry[]

Field	Type	Description
`eventName`	`string`	Event name (e.g. `"order.item_added"`)
`data`	`unknown`	Event payload
`version`	`number`	Stream version at which this event was appended
`timestamp`	`Date`	When the event was recorded

Event Versioning

Automatic Upcasting

Pass an event definition registry to serve() or createWorker() to automatically upcast old events before they reach your handler:

import { defineEvent, createEventDefinitionRegistry } from '@ironflow/core';
import { serve } from '@ironflow/node';

const registry = createEventDefinitionRegistry();
registry.register(defineEvent({ name: 'order.created', version: 1 }));
registry.register(defineEvent({
  name: 'order.created',
  version: 2,
  upcast: (data) => ({ ...data, currency: 'USD' }),
}));

const handler = serve({
  functions: [myFunction],
  eventDefinitions: registry,
});

Handlers automatically receive upcasted event data. See @ironflow/core docs for defineEvent() and createUpcasterRegistry() details.

The registry applies version - 1 → version upcasters in sequence (v1 → v2 → v3). It ignores upcast on version 1. An upcaster must return all fields of the data, so spread the fields that it does not change. Projections do not upcast: eventDefinitions does not reach the projection runner, so a projection handler sees the stored data at the version that was written.

createUpcasterRegistry() in @ironflow/core is a lower-level helper. serve() and createWorker() do not accept it. Call registry.upcast(name, data, fromVersion, toVersion) yourself where you read events.

Webhooks

createWebhook

Declare an inbound webhook source: how to verify a request, and how to turn its payload into an Ironflow event.

import { createWebhook } from '@ironflow/node';

const stripeWebhook = createWebhook({
  id: 'stripe',
  verify: (req) => {
    const sig = req.headers['stripe-signature'];
    return stripe.webhooks.constructEvent(req.body, sig, secret);
  },
  transform: (payload) => ({
    name: `stripe.${payload.type}`,
    data: payload.data.object,
    idempotencyKey: payload.id,
  }),
});

Field	Type	Description
`id`	`string`	Webhook source ID.
`verify`	`(req: WebhookRequest) => unknown \| Promise<unknown>`	Verifies the request. Throw to reject (HTTP 401). The return value is ignored; `transform` receives the parsed body.
`transform`	`(payload: unknown) => WebhookEvent \| Promise<WebhookEvent>`	Maps the payload to the event to emit. A body with content type `application/x-www-form-urlencoded` arrives as an object, with an array for a repeated key. Any other body is parsed as JSON.

Pass the webhook to serve({ functions, webhooks: [stripeWebhook] }). The handler serves a request to a path that contains /webhooks/{id}, for example POST /webhooks/stripe.

For server-side webhook source management (rotate secrets, ingest tokens, deliveries), use client.webhooks — see Admin & Tenant Operations.

Command Idempotency

CommandDedup

Atomic command-level idempotency backed by a NATS KV bucket. tryClaim reserves the command ID before any handler work runs: the winner gets null and proceeds. Later callers receive the stored entry, which can still be the initial pending claim. Only an application completion marker proves that the entry contains a result. The helper does not atomically commit handler effects.

Build one with client.commandDedup() and reuse it across requests — do not call it per request. The bucket is created lazily on the first operation.

type Entry<T> = { status: 'claimed' } | { status: 'done'; result: T };
const dedup = client.commandDedup<Entry<OrderResult>>('order-commands');

const prior = await dedup.tryClaim(commandId, { status: 'claimed' });
if (prior !== null) {
  if (prior.status === 'done') return prior.result;
  throw new Error('Command is pending; reconcile its effects before retrying');
}

let result: OrderResult;
try {
  result = await runHandler(); // Effects must support idempotent retries.
} catch (err) {
  await dedup.release(commandId).catch(() => {}); // Only the claim winner reaches here.
  throw err;
}
await dedup.finalize(commandId, { status: 'done', result });
return result;

A prior entry can be pending. This example rejects pending entries instead of reporting success. Recover them through application-specific reconciliation and idempotent effects. For entity streams, see the CQRS walkthrough.

Member	Signature	Description
`client.commandDedup`	`<T>(bucketName: string, options?: { ttlSeconds?: number }) => CommandDedup<T>`	TTL defaults to `DEFAULT_COMMAND_DEDUP_TTL_SECONDS` (604800 — 7 days). Pass `0` for no expiry. `new CommandDedup(client.kv(), bucketName, ttlSeconds)` is equivalent.
`tryClaim`	`(commandId: string, claim: T, isOwner?: (prior: T) => boolean) => Promise<T \| null>`	Claims the ID. `null` means you won and should run the handler. Retries up to 3 times when the winner releases between your write and read, then throws `IronflowError` with code `COMMAND_DEDUP_RACE`.
`finalize`	`(commandId: string, result: T) => Promise<void>`	Stores the final result under the claim.
`release`	`(commandId: string) => Promise<void>`	Drops the claim so a retry can run.

If the claim write fails with an error other than 412, the claim state is unknown: the write may have committed, and the helper does not retry it. To let a retry recognize its own orphaned claim, put a token that stays the same across retries in the claim and pass isOwner:

const token = messageId; // the same value on every retry of this command
const prior = await dedup.tryClaim(
  commandId,
  { token, status: 'claimed' },
  (p) => p.token === token && p.status === 'claimed',
);

isOwner runs only on a prior entry. It must return false for a finalized result, or a finished command replays. Run one retry lineage at a time: two concurrent callers with the same token both win.

This is distinct from step memoization: step.run protects steps inside one run, while CommandDedup deduplicates a command across separate runs.

Admin & Tenant Operations

The Node client exposes these admin namespaces. apiKeys, orgs, roles, policySet, tenants, schemas, webhooks and consumerGroups also exist on the browser client; see the browser package reference for their method signatures and examples — the shapes are identical. users, projects, environments, secrets, projections, sqlProjections, agentTools, deployments, capacity and debounce are Node-only namespaces.

Namespace	Methods	Purpose
`client.apiKeys`	`create`, `list`, `get`, `delete`, `rotate`	Environment API key management
`client.orgs`	`create`, `list`, `get`, `update`, `delete`	Organization CRUD (admin RBAC)
`client.roles`	`create`, `list`, `get`, `update`, `delete`	RBAC roles (admin RBAC)
`client.policySet`	`get`, `put`, `validate`, `dryRun`, `versions`, `version`, `rollback`	Rego policy set of the organization (admin RBAC)
`client.users`	`create`, `list`, `get`, `update`, `delete`, `changePassword`	Dashboard users (admin RBAC)
`client.tenants`	`list`, `provision`	Tenant listing and provisioning (admin RBAC)
`client.projects`	`list`, `create`, `update`, `delete`	Project resource hierarchy
`client.environments`	`list`, `create`, `update`, `delete`	Environment CRUD
`client.secrets`	`get`, `set`, `update`, `patch`, `list`, `delete`	Programmatic secrets CRUD
`client.schemas`	`register`, `list`, `get`, `getVersion`, `delete`, `testUpcast`, `listHistory`, `getRevision`	Event schema registry
`client.webhooks`	`create`, `getSource`, `listSources`, `updateSource`, `deleteSource`, `rotateSecret`, `expireSecretPrev`, `disableSignatureVerification`, `rotateIngestToken`, `listDeliveries`	Webhook source management
`client.projections`	`get`, `list`, `listPartitions`, `getStatus`, `rebuild`, `getRebuildJob`, `cancelRebuild`, `pause`, `resume`, `delete`, `waitForCatchup`, `waitForCatchupBatch`, `waitForEvent`, `listHistory`, `getRevision`	Projection control plane
`client.sqlProjections`	`create`, `query`	SQL-backed projection helpers
`client.consumerGroups`	`create`, `get`, `list`, `update`, `delete`	Durable pub/sub consumer group definitions
`client.agentTools`	`list`	Agent tools registered with the server (cursor-paginated)
`client.deployments`	`register`, `get`, `list`, `getExecutionRouting`	Immutable executable deployments and function routing
`client.capacity`	`listLanes`, `listQueue`, `listLeases`, `listBuckets`, `listSessions`, `listCredits`, `stats`	Capacity and dispatch views (platform credentials)
`client.debounce`	`list`, `cancel`	Armed debounce timers

Timestamps on these results are Date values. created_at and updated_at keep their names on API keys, organizations, roles, policy sets, projects, environments and secrets. User and Tenant use createdAt and updatedAt. APIKey.expires_at, APIKey.last_used_at and PolicySet.updated_at are optional, absent when unset. An unreadable value throws SchemaValidationError. Its message names the field and prints only the timestamp value.

The admin namespaces are gated by handler RBAC on the calling key, not by a licence tier — Ironflow ships a single build with no Core/Enterprise split (ADR 0015).

const client = createClient();

// Create an environment API key
const key = await client.apiKeys.create({ name: 'ci-runner' });

// Register a new event schema version
const schema = await client.schemas.register({
  name: 'order.placed',
  version: 2,
  schema: { type: 'object', properties: { /* ... */ } },
});

created_at on a schema from schemas.get, getVersion or list is a Date. It is absent on the schemas.register result, because the server answers that call with a status only.

Deployments, definition history, capacity and debounce

Types for these namespaces are in @ironflow/core.

// Executable deployments. register() stages a deployment; it never changes
// which code serves new runs. Needs functions:register. The reads need functions:read.
const { deployment, created } = await client.deployments.register({
  functionId: 'process-order',
  deploymentId: '2026-10-16-a',
  artifactDigest: `sha256:${digestHex}`, // 64 lowercase hex digits
  snapshot: { triggers: [{ event: 'order.placed' }], timeoutMs: 30000 },
  mode: 'pull', // 'push' also needs endpointUrl
});
// created is false when the identical deployment already existed.
// A different artifact, snapshot, mode or endpoint under an existing
// deploymentId fails with already_exists.

const one = await client.deployments.get('process-order', '2026-10-16-a');
const all = await client.deployments.list('process-order', { limit: 20 }); // newest first
const routing = await client.deployments.getExecutionRouting('process-order');
// routing.mode: 'legacy' | 'adopting' | 'versioned'
// deployment.createdAt is a Date. routing.updatedAt and routing.transitionStartedAt
// are optional Dates: updatedAt is absent while the function has no routing row.

// Definition history: definitions archived before a replacement or deletion.
// Newest first. The current definition is not in the history.
let page = await client.projections.listHistory('order-totals', { limit: 20 });
while (page.hasMore) {
  page = await client.projections.listHistory('order-totals', {
    limit: 20,
    beforeRevision: page.nextBeforeRevision, // exclusive cursor; 0 means no more pages
  });
}
const rev = await client.projections.getRevision('order-totals', 3);
// Schema history is per (event name, version).
const schemaPage = await client.schemas.listHistory('order.placed', 2, { limit: 20 });
const schemaRev = await client.schemas.getRevision('order.placed', 2, 1);

// Capacity views. They need platform credentials (an ifplatform_ key or a
// platform session). A tenant key gets 403.
const queue = await client.capacity.listQueue({ env: 'env_default', limit: 50 });
const stats = await client.capacity.stats();

// Debounce timers. cancel() takes the raw key and base64url-encodes it.
const entries = await client.debounce.list();
await client.debounce.cancel('env_default', 'process-search', 'user-42');

limit on a history call defaults to 50, and the maximum is 200. The capacity methods take { env?, function?, limit? }. The leases and credits views ignore function. The leases view always returns an empty function_id, so match a lease to its run by run_id. Capacity and debounce rows keep the server’s snake_case field names. Deployment and history results use camelCase. Their timestamps are Date values, and an unreadable one throws SchemaValidationError. last_dispatched_at on a bucket and reservation_expires_at on a queue entry are optional, absent until the bucket first dispatches and until the entry holds a reservation. See Executable Deployments for the deployment fields, errors and permissions.

Server Introspection

listFunctions

List registered functions:

const functions = await client.listFunctions();
for (const fn of functions) {
  console.log(fn);
}

Each entry’s created_at, updated_at and next_run are Date values. next_run is absent unless the function has a schedule.

listWorkers

List the HTTP pull workers registered in the environment. In a cluster the list covers every node, and active_jobs counts the job leases each worker holds:

const workers = await client.listWorkers();
for (const w of workers) {
  console.log(w);
}

getCapabilities

Discover what the server supports:

const caps = await client.getCapabilities();
console.log(caps.transports, caps.features, caps.version);

health

Check server health:

const status = await client.health();
console.log('Server status:', status); // "healthy"

ready

health() is liveness. ready() reads GET /ready, which checks the database, NATS and the engine streams. A server that is not ready makes it throw an IronflowError with code NOT_READY, status 503 and the failing dependencies in details.issues, such as stream:PUBSUB. It makes one attempt and does not retry.

const { status, warnings } = await client.ready();
console.log(status); // "ready"

Error Handling

Global Error Handler

Use the onError option on createClient() to observe all client errors in one place. See Global Error Handler (onError) in the Client section above.

Built-in Errors

The full set @ironflow/node re-exports from its root entry point:

import {
  IronflowError,          // base class: code, retryable, details
  StepError,              // step failure; has stepId, stepName
  TimeoutError,
  ValidationError,
  SchemaValidationError,
  SignatureError,
  FunctionNotFoundError,
  RunNotFoundError,
  NonRetryableError,
  RetryAfterError,        // retryable; asks for a delay: (message, retryAfterMs)
  UnauthenticatedError,   // HTTP 401
  UnauthorizedError,      // HTTP 403
  EnterpriseRequiredError,// HTTP 402, legacy (ADR 0015)
  ConflictError,          // HTTP 409, not a lost race — wait, do not retry
  ContendedError,         // HTTP 409 from a lost CAS race — re-read, then reissue
  PreconditionFailedError,   // HTTP 412, If-Match / If-None-Match failed (files)
  PreconditionRequiredError, // HTTP 428, the write needs If-Match (policy set)
  PayloadTooLargeError,      // HTTP 413 (files)
  UnsupportedMediaTypeError, // HTTP 415 (files)
  POLICY_SET_ERROR_CODES,    // constant: the `code` values of policy set errors
  isRetryable,
  isIronflowError,
} from '@ironflow/node';

IronflowError.code is the lowercase Connect code (not_found, invalid_argument, unauthenticated, permission_denied, already_exists, internal, …) on every call, whether the route is REST or Connect. A domain code, such as BUCKET_NOT_EMPTY, keeps its own name. A server reply with no code gets the Connect code of its HTTP status. UnauthenticatedError, UnauthorizedError and ConflictError carry unauthenticated, permission_denied and already_exists.

StreamingUnsupportedError also ships from the root entry point — see createProjectionRunner. Core defines more classes than Node re-exports (StepTimeoutError, RunWaitTimeoutError, RunFailedError, RunCancelledError among them); import those from @ironflow/core directly.

import { IronflowError, StepError, isRetryable } from '@ironflow/node';

try {
  await step.run('risky-operation', async () => {
    // ...
  });
} catch (error) {
  if (error instanceof StepError) {
    console.log('Step failed:', error.stepId);
  }

  if (isRetryable(error)) {
    // Will be automatically retried
  }
}

Non-Retryable Errors

Mark errors as non-retryable to prevent automatic retries:

import { NonRetryableError } from '@ironflow/node';

await step.run('validate', async () => {
  if (!isValid(data)) {
    throw new NonRetryableError('Invalid data - do not retry');
  }
  return data;
});

Agent limits and persistence

The @ironflow/node/agent helpers wrap durable step primitives. Your application or agent framework still owns reasoning and provider calls.

agent, defineTool, exposeMcp, llm, memory, spawn and approve are Ready: the contract is settled and no breaking change is planned. The same holds for the Go and Python agent APIs.

Environment

ctx.run.environment is the environment of the run’s outbound calls. A pull worker sets it to the value it polls with (environment, else IRONFLOW_ENV, else unset). serve() sets it to environment, else IRONFLOW_ENV, else leaves it unset. Agent memory and step.publish send it as X-Ironflow-Environment-ID. For a client built in a handler, pass createClient({ environment: ctx.run.environment }). exposeMcp({ environment }) scopes tool registration; the default is IRONFLOW_ENV, else no header.

Logical turns and provider limits

maxTurns limits logical llm.complete calls in a handler, with a default of 20. The counter advances even when a completion is returned from memoization during replay. It is not a count of provider requests, a token limit, or a cost budget. Provider retries can make several requests for one logical turn.

Set provider options such as max_tokens in the actual application/provider SDK call inside call. The wrapper invokes that closure; it does not forward messages, model, or other request hints to a provider on your behalf. Keep reasoning, prompts, and provider configuration in your application or chosen agent framework. Token and cost budgets are not shipped agent controls. Returned usage metadata is accounting information, not hard spending enforcement: provider calls and retries can incur charges before a result is persisted.

Audit recording and durable results

recording: false disables optional function audit events. It does not disable durable step-result persistence. llm.complete memoizes the completion returned by the provider closure, including any returned metadata. Return only the data needed for replay; omit unnecessary raw responses and sensitive content before returning from call.

Keep execution audit, durable step results, application case state, and projection-backed memory distinct. Audit events describe execution; step results drive replay; case state holds business facts; memory reads a projection over explicitly appended events. Raw audit logs are not automatically appropriate or authorized model context. Applications must curate knowledge, redact sensitive data, and enforce authorization; a projection does not supply those policies automatically.

Global pruning applies to every function’s audit rows; audit retention does not govern all execution data. See audit and agent data boundaries.

Agent Errors

The @ironflow/node/agent subpath exports nine of its own, all extending IronflowError:

Error	When
`ToolNotFoundError`	`tool(name, args)` called with a name not in `config.tools`.
`ToolValidationError`	Tool input failed its Zod schema.
`DuplicateToolError`	Two tools registered under the same `name`.
`LLMError`	Base LLM failure; `LLMInvalidJSONError`, `LLMMaxTokensError` and `LLMRefusalError` are subclasses.
`MaxTurnsExceededError`	Agent exceeded its logical `maxTurns` limit (default 20).
`MemoryProjectionRequiredError`	`agent()` was given a `memory` config with an empty `projection`.

import { ToolValidationError } from '@ironflow/node/agent';

Utilities

Duration Parsing

import { parseDuration } from '@ironflow/node';

parseDuration('30s');  // 30000 (ms)
parseDuration('5m');   // 300000
parseDuration('2h');   // 7200000
parseDuration('1d');   // 86400000

ID Generation

import { generateId, createRunId, createStepId } from '@ironflow/node';

const id = generateId();                  // 'm1abc23-x4y5z6' (no arguments)
const runId = createRunId('run_abc123');   // Branded RunId type
const stepId = createStepId('step_def');  // Branded StepId type

Logging

import { createLogger, createNoopLogger } from '@ironflow/node';

const logger = createLogger({
  prefix: '[myapp]',
  level: 'debug',  // 'debug' | 'info' | 'warn' | 'error' | 'silent'
});

logger.info('Starting...');
logger.debug('Debug data', { foo: 'bar' });

// Disable logging
const noopLogger = createNoopLogger();

Correlated application logs

ctx.logger adds five reserved fields to each call in push, polling, and streaming execution. run_id is the full dispatch run ID, function_id is the dispatch function ID, and attempt is the engine run attempt. step_id is the allocated memoization key inside an active callback, including escaped names and branch scope, or an empty string outside it. trace_id is the valid current execution trace ID, including unsampled traces, or an empty string without one. Step correlation works without OpenTelemetry installed.

The default console output prints all five fields and the full run ID prefix. Pass a Logger as userLogger to serve(), createWorker(), or createStreamingWorker() to use your application’s sink. Its filtering, formatting, exceptions, and silence remain yours. userLogger: false silences execution logs. The existing logger option controls transport diagnostics. Execution-step diagnostics still share ctx.logger with callback logs and keep their existing messages and levels. Application attributes survive, with the five reserved fields taking precedence over collisions.

const worker = createWorker({
  functions: [processOrder],
  userLogger: applicationLogger,
});
// In processOrder's callback:
// logger.info('charged', { orderId: event.data.orderId });

For an explicit OTel Logs route, install your application’s Logs SDK and the optional @opentelemetry/api peer. Obtain a logger from your provider and adapt it once. Tested versions are @opentelemetry/api 1.9.0 and @opentelemetry/api-logs and @opentelemetry/sdk-logs 0.220.0 on Node.js 22+.

import { createOTelLogger, createWorker } from '@ironflow/node';
import { LoggerProvider, SimpleLogRecordProcessor } from '@opentelemetry/sdk-logs';

// exporter is created and managed by your application.
const provider = new LoggerProvider({
  processors: [new SimpleLogRecordProcessor({ exporter })],
});
const worker = createWorker({
  functions: [processOrder],
  userLogger: createOTelLogger(provider.getLogger('my-app')),
});

The adapter emits once per call, with the original message body, mapped DEBUG/INFO/WARN/ERROR severity, application attributes, and explicit current OTel context for intrinsic trace/span correlation. It works without a global context manager. It has no console tee and imports no Logs package. Application providers control filtering, limits, timestamps after processing, and export. Use OTel-compatible attribute values. Explicit Error attributes become { name, message } without automatically including stacks. There is no implicit exception capture. Native custom sinks receive explicit errors unchanged.

Applications own provider setup, exporters, flush, shutdown, and destinations. Configuring tracing alone does not enable Logs export. Ironflow introduces no log collection, storage, network export, or provider setup. Do not bridge an already bridged sink a second time for the same destination.

Memoized callbacks do not run or log again. Other callback logs can repeat on resume. Logs are not checkpointed or deduplicated. Detached async work retains its inherited trace context but loses the step label after callback exit, cancellation, or timeout, even if user code keeps running. Arbitrary console calls and independently created loggers are not enriched.

Run Context

The Connect calls that an IronflowClient makes from inside a function handler (emit, streams.append, and so on) send the run ID in X-Ironflow-Run-ID. This ties the events and entity streams that the handler writes to its run. serve() and the workers set the run context for you. getCurrentRunId() returns the current run ID, or undefined outside a run. withRunContext(runId, fn) runs fn in a run context, for tests and custom instrumentation.

import { getCurrentRunId, withRunContext } from '@ironflow/node';

withRunContext('run_abc123', () => {
  console.log(getCurrentRunId()); // 'run_abc123'
});

Constants

import {
  DEFAULT_SERVER_URL,   // 'http://localhost:9123'
  DEFAULT_PORT,         // 9123
  DEFAULT_HOST,         // 'localhost'
  DEFAULT_TIMEOUTS,     // { CLIENT, FUNCTION, TRIGGER_SYNC, INVOKE_FUNCTION_SYNC, SYNC_TRANSPORT_HEADROOM }
  DEFAULT_RETRY,        // { MAX_ATTEMPTS, INITIAL_DELAY_MS, BACKOFF_FACTOR, MAX_DELAY_MS }
  DEFAULT_WORKER,       // { MAX_CONCURRENT_JOBS, HEARTBEAT_INTERVAL_MS, RECONNECT_DELAY_MS, CHECKPOINT_INTERVAL_MS, ... }
  ENV_VARS,             // Environment variable names
} from '@ironflow/node';

Environment Variables

Variable	Description	Default
`IRONFLOW_SERVER_URL`	Server URL	`http://localhost:9123`
`IRONFLOW_SIGNING_KEY`	The Node SDK does not read this variable. Your code reads it and passes it to `serve` as `signingKey`. With no key, the handler does not verify requests.	-
`IRONFLOW_API_KEY`	API key for authentication	-
`IRONFLOW_LOG_LEVEL`	Log level	`info`
`IRONFLOW_ENV`	Target environment ID for `serve()` and `createWorker()`	-
`IRONFLOW_NO_DISCOVERY`	Set to `1` to turn off the engine discovery file (ADR 0098).	-

When none of IRONFLOW_SERVER_URL and IRONFLOW_API_KEY is set, the SDK reads the nearest .ironflow/engine.json (loopback http URLs only) to fill the unset variables. Environment values always win.

Testing

The @ironflow/node/test entry point provides createTestClient for running a function without a server. mockStep(name, fn) replaces a step.run body, mockInvoke(functionId, fn) replaces step.invoke / step.invokeAsync, and sendEvent(eventName, data) queues an event for step.waitForEvent. emit(eventName, data) runs the function that has a trigger for the event and resolves to a TestRun with status, output, error, steps, compensationsRan and stepOutput(name). Every step.run and step.invoke call needs a mock; an unmocked step throws. An unqueued wait with onTimeout: "return-null" resolves immediately to null; a default wait still fails for a missing fixture. sendEvent(name, null) queues a real event whose data is null.

import { createTestClient } from '@ironflow/node/test';

const tc = createTestClient({ functions: [processOrder] });
tc.mockStep('validate', () => ({ valid: true }));

const run = await tc.emit('order.placed', { orderId: '123' });
console.log(run.status, run.stepOutput('validate'));

See the Testing Workflows guide for waiting, invoke and saga patterns.

OpenTelemetry tracing

Install the optional @opentelemetry/api peer (^1.9.0) and configure your application’s OpenTelemetry tracer provider and context manager before starting workers or serve(). The SDK uses the API only. It does not install an exporter or configure global tracing. Node.js 22 or later is required.

REST polling and streaming workers continue each job’s W3C traceparent and tracestate. Push handlers read these fields from HTTP headers. Each execution creates a process consumer span, and each executed step.run creates a child step.run span. Memoized steps do not create spans. Parallel branches retain their execution ancestry. A poll request’s ambient span never becomes the job’s parent. Invalid parents start a new trace; valid unsampled parents still propagate.

Failures set error status. Yields and worker cancellation end spans without error status. Spans contain no baggage, payloads, secrets, or exception messages. Step names are recorded as ironflow.step.name; avoid putting sensitive data in step names. Span names and error events use bounded SDK labels.

Without the optional peer, import and execution continue normally without spans. Use your application’s OpenTelemetry setup to select sampling and exporters.

Requirements

Node.js 22+

Complete Example

import { ironflow, serve, createClient } from '@ironflow/node';

// Define a function
const processOrder = ironflow.createFunction(
  {
    id: 'process-order',
    name: 'Process Order',
    triggers: [{ event: 'order.placed' }],
  },
  async ({ event, step }) => {
    const order = event.data as { orderId: string; amount: number };

    // Step 1: Validate
    await step.run('validate', async () => {
      if (order.amount <= 0) {
        throw new Error('Invalid amount');
      }
      return { valid: true };
    });

    // Step 2: Sleep
    await step.sleep('delay', '5s');

    // Step 3: Process
    const result = await step.run('process', async () => {
      return {
        orderId: order.orderId,
        status: 'processed',
      };
    });

    return result;
  }
);

// Register function and emit events from server-side code
const client = createClient();
await client.registerFunction({
  id: 'process-order',
  triggers: [{ event: 'order.placed' }],
  endpointUrl: 'http://localhost:3001/api/ironflow',
  preferredMode: 'push',
});

// Serve as push mode handler
export const POST = serve({
  functions: [processOrder],
});