Engine API

@flow-state-dev/engine — Action runtime, stores, SSE streaming, orchestration.

Runtime

createFlowState(options)

The main entry point. Assemble a runtime from declarative config. Returns a FlowState handle. Most users want this rather than the lower-level registry and router below.

import { createFlowState, inMemoryStores } from "@flow-state-dev/engine";

const flowstate = createFlowState({
  flows: { myFlow },
  models: { default: "openai/gpt-6-luna" },
  stores: { default: { primary: inMemoryStores() } },
});

flows and stores are required; the rest have defaults. Every option, with its type and default, is catalogued in Runtime options — including the model, store, worker, durability, and sweeper groups.

Construction is synchronous and validates config (empty stores, unknown defaultProfile) by throwing FlowStateConfigError. Stores initialize lazily on first getRouter() / ready().

FlowState

The handle returned by createFlowState.

interface FlowState<TSettings extends object = FlowStateSettings> {
  /** Resolve the lazy router. First call triggers async store init. */
  getRouter(): Promise<FlowApiRouter>;
  /** Eager warmup. Idempotent. */
  ready(): Promise<void>;
  /** Drain in-process background work, close the worker, release pooled resources. */
  dispose(): Promise<void>;
  /** Admit one flow after startup. Throws if it is refused. */
  register(flow: FlowInstance): void;
  /** Release one address. False if nothing was registered under that id. */
  unregister(id: string): boolean;
  /** The active profile name. */
  readonly activeProfile: string;
  /** The settings bag, typed via TSettings. */
  readonly settings: TSettings;
  /** Diagnostics. */
  readonly meta: {
    /** The instance ids currently registered. */
    flowKeys: string[];
    profileKeys: string[];
    declaredSlots: Record<string, CapabilitySlot[]>;
  };
}

Registering a flow after startup

createFlowState({ flows }) takes the flows an app knows about when it starts. register adds one later, and unregister removes one.

Both run the same checks construction runs, so a duplicate id or a conflicting scope schema is refused and nothing changes. register takes one flow rather than a list, because admitting a batch would have to either roll the whole batch back on a refusal or leave the earlier entries admitted — and a caller registering several almost always wants to keep the rest and know which one was refused.

The registry is read once per request, so a flow registered here is served from the next request onward, in this process. Other processes pick it up when they next start. A request already running is unaffected by an unregister: it holds the flow instance it resolved, so its stream is not truncated and what it wrote is not discarded.

Two things go stale. Anything that caches the flow list misses later registrations, so read it per request. And an adapter that validates flows when it starts — the webhook adapter checks that every declared provider is configured — has already run, so a flow registered afterwards is not checked until the next start.

Shutdown

dispose() runs in order:

1. Waits for background work still running in this process. A job handed to a queue is not waited for here — but if this process also consumes that queue, step 5 waits for whatever it has already claimed.
2. Bounds that wait with dispatchDrainTimeoutMs, default 30000 ms. It's a ceiling, not a target: work that finishes sooner is not delayed. 0 means don't wait at all.
3. Cancels whatever is still running when the budget runs out, and gives it a brief window, inside that same budget rather than added to it, to unwind.
4. Reports the request ids and session ids it gave up on, on stderr. That report prints even when the runtime's logger is silenced, since work may have been left unfinished.
5. Closes the worker and releases pooled resources across every declared store adapter. Closing the worker waits for any queue job this process has already claimed, and that wait is not bounded by dispatchDrainTimeoutMs — it takes as long as the job does. Size your platform's kill timeout for the longest job.

Shutdown mostly does not write a terminal status on background work's behalf. It cancels the work rather than marking those records finished or failed. One case doesn't follow that yet: work still waiting behind a concurrency limit when shutdown reaches it is recorded aborted without ever having started. For what the cancelled work leaves in the task board and the request log, and how each recovers, see What a stopped process leaves behind.

fsdev run and fsdev chat shut down through the same path. See Waiting for in-process work.

StoreAdapter

A tagged value a store-adapter factory returns. Declares which capability slots it backs and realizes them on demand.

interface StoreAdapter {
  /** Slots this adapter is willing to back. */
  readonly capabilities: ReadonlyArray<CapabilitySlot>;
  /** Realize the adapter for the given slots. Opens pools / runs schema init. */
  resolve(slots: ReadonlyArray<CapabilitySlot>): Promise<Partial<StoreRegistry>>;
  /** Release pooled resources. Called by FlowState.dispose(). */
  dispose?(): Promise<void> | void;
}

CapabilitySlot is "primary" | "blobs" | "queue" | "scheduler". Only primary is implemented in Phase 1; the others are declared but forward-compatible.

Store-adapter factories: inMemoryStores(), filesystemStores({ rootDir }) (this package), postgresStores(options) (@flow-state-dev/store-postgres), sqliteStores(options) (@flow-state-dev/store-sqlite), vercelPostgresStores() (@flow-state-dev/vercel/store).

Registry & Router (lower-level)

createFlowState wraps these. Reach for them only for custom transports or advanced wiring.

createFlowRegistry()

Create a registry for flow instances. The registry indexes every instance by its id, which is what every route, CLI target, and queue job carries.

import { createFlowRegistry } from "@flow-state-dev/engine";

const registry = createFlowRegistry();
registry.register(myFlow);          // singleton: id is its kind
registry.registerMany([east, west]); // collection instances: their own ids

registry.get("my-flow");     // the singleton
registry.get("review-east"); // one collection instance
registry.get("review");      // undefined: a collection's kind is not an address
registry.list();             // every registered instance

get(id) is an exact lookup. There is no first-registered fallback, and a miss is undefined. Ids are unique across the registry regardless of kind: a second register with an id already taken throws FlowIdentityConflictError (reason: "duplicate-id"), as does a singleton registered under a custom id ("singleton-id-mismatch") or a collection instance without one. A kind cannot mix a singleton with collection instances.

GET /api/flows lists what is registered, one entry per instance: { id, kind, cardinality, actions, ... }. createFlowState builds this registry for you from its flows option; reach for it directly only when wiring a custom host.

createFlowApiRouter(options)

Create HTTP route handlers for the flow API from a registry.

import { createFlowApiRouter } from "@flow-state-dev/engine";

const router = createFlowApiRouter({
  registry,
  stores,                                // a resolved StoreRegistry
  modelResolver: createModelResolver(),  // optional
  voiceProvider: new OpenAIVoiceProvider({ apiKey: process.env.OPENAI_API_KEY }), // optional, for TTS/STT
});

export const { GET, POST, PATCH, DELETE } = router;

Pass maxRecordedValueBytes to change the largest block output or tool result the runtime records in an item, in serialized bytes. The default is 256 KiB. Larger values are recorded as a placeholder; see Values too large to record.

const router = createFlowApiRouter({
  registry,
  maxRecordedValueBytes: 1024 * 1024, // 1 MiB
});

Raising it keeps more in your store and in model history.

parseFlowRoute(method, path)

Parse a flow API path into its components (the addressed flow id, action, sessionId, and so on). The first segment of an action or stream route is an instance id, which for a singleton is its kind. See Addressing an instance.

path is an array of path segments beneath the mount prefix, each already percent-decoded once. A framework catch-all such as Next's params.path hands over that shape. parseFlowRoute decodes nothing further, so a segment that still contains %5F or / after that one decode is read as written.

decodePathSegments(rawPath)

Splits a raw, still-encoded URL path (the part beneath the mount prefix) on / and decodes each segment once. The result is the array parseFlowRoute takes. Use it in a host that only has the raw URL. A segment with invalid percent-encoding is kept as written.

Execution

createExecutionContext(options)

Create a block execution context manually (for advanced use).

runAction(options)

Execute a flow action programmatically. The flow you pass owns every record the run creates, and a sessionId that belongs to another instance is refused before the action starts. See Manual flow execution.

executeBlock(block, input, ctx)

Execute a block directly with a given context.

Stores

createFilesystemStores(options?)

Filesystem-backed persistence for local development. Durable across restarts, but its per-request event log doesn't scale under production load — use SQLite (single server) or Postgres (multi-instance) in production.

import { createFilesystemStores } from "@flow-state-dev/engine";

const stores = createFilesystemStores({ rootDir: ".fsdev/data" });

createInMemoryStores()

In-memory persistence for testing.

import { createInMemoryStores } from "@flow-state-dev/engine";

const stores = createInMemoryStores();

Model Resolution

createModelResolver(options?)

Create a model resolver. Auto-detects providers from environment variables with zero config, or accepts explicit keys, intents, and retry policy.

import { createModelResolver } from "@flow-state-dev/core/models";

// Zero-config: auto-detects from env vars
const resolver = createModelResolver();

// With options:
const resolver = createModelResolver({
  keys: { openai: "sk-..." },
  defaultModel: "openai/gpt-6-luna",
  intents: { utility: ["openai/gpt-6-luna", "anthropic/claude-haiku-5-5"] },
  retryPolicy: { maxAttemptsPerModel: 3 },
});

Model strings look like "openai/gpt-6.1-sol" or "anthropic/claude-sonnet-5-5". To name the gateway that serves a model, put it before a colon: "vercel:openai/gpt-6.1-sol".

parseModelString(modelString)

Split a model string by lane, using the string alone:

parseModelString("openai/gpt-6.1-sol");          // { type: "direct", provider: "openai", modelId: "gpt-6.1-sol" }
parseModelString("vercel:openai/gpt-6.1-sol");   // { type: "alias", alias: "vercel", modelId: "openai/gpt-6.1-sol" }
parseModelString("intent/chat");             // { type: "intent", intentName: "chat" }

An alias result doesn't say whether the name is a gateway or a provider. The resolver decides that from its providers and gateways. A string with three slash-separated parts throws, naming its colon spelling.

Voice

Voice surfaces are owned by a VoiceProvider passed as voiceProvider to createFlowApiRouter (or voice: { provider } on createFlowState). A provider declares its abilities and implements speak / speakStream / transcribe / listVoices. Concrete providers ship in their own packages (e.g. @flow-state-dev/voice-openai). See Voice for the full guide.

createCompositeVoiceProvider(config)

Combine providers per ability (e.g. synthesize with one, transcribe with another). Exported from @flow-state-dev/core.

createSentenceBuffer()

Create a sentence-boundary detection buffer for TTS text chunking.

createTTSPipeline(options)

Create the synthesis pipeline that converts text deltas into OutputAudioContent.

createTTSEmitterHook(options)

Create an event observer that wires the TTS pipeline to a ResponseEmitter.

Streaming

createResponseEmitter(options)

Create a stream emitter for manual event emission.

encodeStreamEvent(event)

Encode an event for SSE transmission.

serializeSSEFrame(event)

Serialize an event into SSE wire format.

replayRequestEvents(options)

Replay stored events from a completed request (for resume).

Errors

FlowError

Base error class with code, retryable, blockName, scope, and cause.

Subclasses:

• ValidationError — Schema validation failure (not retryable)
• NetworkError — Network issues (retryable)
• TimeoutError — Operation timeout (retryable)
• RateLimitError — Rate limit hit (retryable)
• ModelError — Model provider error (retryable)
• ToolExecutionError — Tool block failure (varies)

normalizeError(error)

Convert any thrown value to a FlowError.