@flow-state-dev/engine

The runtime. Register flows, execute actions, stream results — one config object to a complete API.

Installation

pnpm add @flow-state-dev/engine

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

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

Mount it with a platform adapter (@flow-state-dev/vercel/next on Vercel, @flow-state-dev/next elsewhere):

import { flowstate } from "@/lib/flowstate";
import { createVercelNextHandler } from "@flow-state-dev/vercel/next";

export const { GET, POST, PATCH, DELETE } = createVercelNextHandler(flowstate);
export const runtime = "nodejs";
export const maxDuration = 300;
export const dynamic = "force-dynamic";

That's a full API with action execution, session management, SSE streaming with resume, and state snapshots.

createFlowState

createFlowState(options) builds the runtime from one declarative object and returns a FlowState handle:

• getRouter(): Promise<FlowApiRouter> — resolve the route handlers (first call triggers store init).
• ready(): Promise<void> — eager warmup, idempotent.
• dispose(): Promise<void> — drain in-process background work, close the worker, release pooled resources.
• register(flow) / unregister(id) — add or remove one flow after startup.
• resolveInProcessPrincipal({ flowKind, action, input, userId }) — who an in-process caller is, answered through the same resolvers and organization rules as the HTTP routes. The resolver sees a frozen context with source: "cli", no request, and metadata.body shaped like an HTTP action body ({ userId, input }); fsdev run and fsdev chat use it. Resolves { userId, orgId, from: "resolver" | "development-default" }, or rejects with the resolver's refusal. For a bare registry with no FlowState, call the exported resolveInProcessPrincipal({ registry }, question). source: "cli" is reserved: every inbound host refuses it on a context this entry point did not build.
• activeProfile, settings, meta — read-only diagnostics.

Construction is synchronous; stores initialize lazily and memoized on the first getRouter() / ready(). There's no top-level await, so the same instance works in a Next.js Route Handler.

Registering a flow after startup

createFlowState({ flows }) takes the flows an app knows about when it starts. To add one later, register it:

flowstate.register(flow);      // one instance at a time
flowstate.unregister(flow.id); // returns false if nothing was registered under that id

Registration runs the same checks construction does: a duplicate id is refused, and so is a flow whose user- or org-scoped schemas conflict with one already registered. A refused registration changes nothing.

register takes one flow rather than a list on purpose. 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 flows almost always wants to know which one was refused and keep the rest. Loop, and handle each refusal where it happens.

The registry is read once per request, so a flow registered here is served from the next request onward, in this process. A request already running is unaffected either way: it holds the flow instance it resolved, so unregistering does not cancel it, shorten its stream, or discard what it wrote. A request that arrives after an unregister gets the same answer it would in a process that never had the flow.

Two things go stale, and both are worth knowing. Anything that reads the flow list and caches it will miss later registrations, so read 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.

meta.flowKeys reads the registry, so it lists the ids that are actually being served — including one registered after startup, and not one that has been unregistered.

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, a createFlowState option defaulting to 30000 ms. It's a ceiling, not a target: work that finishes sooner is not delayed. 0 means don't wait at all. Passing detachedDrainTimeoutMs throws — the option was renamed, and honouring it silently would leave the drain on its default.
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. The exception is a run still waiting behind a concurrency limit when the drain reaches it, which is recorded aborted without ever having started. Otherwise the task is taken back by the next claim once its lease has lapsed — the row stays in_progress as a fresh attempt, and settles errored only past the abandonment allowance — and the request record reads interrupted: written by the run itself if it unwinds inside the drain's budget, or by a later runtime start's sweep once its heartbeat has been quiet longer than the staleness threshold. See what a stopped process leaves behind.

Stores and capability profiles

stores is a map of named profiles. A profile maps capability slots (typed containers for a category of storage) to adapters. The required slot is primary — the catch-all state store for sessions, requests, users, orgs, active requests, checkpoints, content, and traces. The blobs, queue, and scheduler slots are declared but forward-compatible; no backing store ships for them yet.

The bus slot carries change notices: every request and session write publishes one on its session's topic, and the session stream wakes on it (see Change bus). Leave it out and the primary backs it when the primary's adapter declares bus, as postgresStores does (its bus crosses processes over LISTEN/NOTIFY); otherwise the bus is in memory and reaches this process only. An adapter named in the bus slot that doesn't declare it fails at ready().

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

createFlowState({
  flows: { myFlow },
  stores: {
    prod: { primary: vercelPostgresStores() },
    dev: { primary: inMemoryStores() },
  },
  defaultProfile: "dev",
});

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).

Profile selection

The active profile resolves on first use, first match wins: process.env.FSD_ENV → defaultProfile → first declared profile. NODE_ENV is intentionally not consulted — an explicit selector keeps a production build from silently pointing at production infrastructure.

Settings

settings is instance-level config blocks read via ctx.settings. Type it by declaration-merging into FlowStateSettings:

declare module "@flow-state-dev/core" {
  interface FlowStateSettings {
    sandbox: { type: "local" | "vercel" | "memory" };
  }
}

createFlowState({
  flows: { myFlow },
  stores: { default: { primary: inMemoryStores() } },
  settings: { sandbox: { type: "local" } },
});

Then read const s = ctx.settings.sandbox inside any block.

Serverless background work

On platforms that freeze the function after the response (Vercel), pass the platform keep-alive primitive at construction so fire-and-forget work isn't killed:

import { after } from "next/server";

createFlowState({ /* ... */ onBackgroundWork: (p) => after(() => p) });

It's a createFlowState option, not a handler option, because the router is built inside createFlowState.

Error capture

errorCapture is an opt-in, block-aware sink for routing runtime block failures to a projected observability service (Sentry, Datadog, Bugsnag). It's distinct from onError, which is an HTTP-level sink. The callback receives a provider-neutral ErrorCaptureEvent (the normalized FlowError plus the failing block's identity and the flow/request/session/user IDs), fires once per failing block, and is fire-and-forget — a throw or rejection is swallowed and logged, never affecting the request.

import * as Sentry from "@sentry/node";

createFlowState({
  /* ... */
  errorCapture: (event) =>
    Sentry.captureException(event.error, {
      user: { id: event.userId },
      tags: { flow: event.flowKind, block: event.blockName ?? "unknown" },
    }),
});

A request that settles failed sends one more event, marked final: true with scope: "request". It comes after rescues and retries had their chance and after the failed record is written, so a failure a retry healed never gets one. Use it when you want one event per failed request (filing a bug, paging someone). If you forward every block failure, as above, skip it with if (event.final) return; so the same failure isn't sent twice.

See the Error capture docs for the full event shape and filtering guidance.

Connection resilience

createFlowState forwards the SSE heartbeat and stale-request sweeper knobs to the router: defaultSseHeartbeatMs, staleSweepIntervalMs, staleSweepThresholdMs, and queuedGraceMs. The defaults suit typical Vercel/Next.js deployments. See the connection resilience guide for tuning.

It also forwards publicReentrySources — the sources your own inbound transports stamp that retry / continue / resume may re-enter. See Inbound transports.

maxChildSessionListLimit sets the largest limit the /children listing route accepts, defaulting to 100. Raise it when conversations run more background work than that: the list is all-time history, so any fixed ceiling eventually hides the oldest finished work. Raise it deliberately — each row resolves its status from the request store and clients re-read this list on every interaction, so a larger ceiling costs more on every turn.

DevTool connection (dev-only)

devtool?: { userId?, bearerToken? } declares how fsdev dev should connect the DevTool UI to this app. userId is the session identity DevTool acts as; bearerToken is sent as Authorization: Bearer on every flow request, so a bearer-gated flow (one whose resolvePrincipal validates a shared secret) is debuggable through DevTool using its real authentication — no bypass.

createFlowState({
  flows: { myFlow },
  stores: { default: { primary: inMemoryStores() } },
  devtool: { userId: "owner", bearerToken: process.env.MY_FLOW_SECRET },
});

fsdev dev reads this off the sync meta.devtool getter (no store init) and injects it into the loopback DevTool page. It is dev-only: the token is exposed only to the loopback page fsdev dev serves, and production serve/deploy paths ignore it. The config type is exported as DevToolConnectionConfig. See the DevTool setup guide.

Lower-level: registry and router

createFlowApiRouter and createFlowRegistry still exist for custom transports and advanced wiring. Most users want createFlowState. The sections below document the lower-level surface.

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

const registry = createFlowRegistry();
registry.register(myFlow);
const router = createFlowApiRouter({ registry, stores });

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

Calling the router in process

createInProcessFetch(app, { basePath? }) takes a FlowState or a router and returns a fetch-shaped function, (input, init?) => Promise<Response>, that hands each request to the engine in this process, with no web server. Its usual home is a client's fetcher:

import { createFlowState, createInProcessFetch } from "@flow-state-dev/engine";
import { createClient } from "@flow-state-dev/client";

const flowState = createFlowState({ flows: { support } });
const fsd = createClient({ userId: "u1", fetcher: createInProcessFetch(flowState) });
await fsd.flow("support").send("reply", { text: "Hi" });

You can also call it directly, as you would fetch:

const fetcher = createInProcessFetch(flowState);
const res = await fetcher("/api/flows/support/actions/reply", {
  method: "POST",
  headers: { "content-type": "application/json" },
  body: JSON.stringify({ userId: "u1", input: { text: "Hi" } })
});

• Only the URL's path is read. The origin is ignored, and a relative path works.
• basePath is the mount, "/api/flows" by default. A path outside it answers 404.
• Each call waits for store init. A failed init answers 500 { error: "Server failed to initialize" }.
• GET, POST, PATCH and DELETE reach the router. Any other method, HEAD included, answers 405.
• SSE bodies stream, and an AbortSignal passed in init reaches the route.

The routes, checks and answers are the same as over HTTP. Credentials ride the request headers, and being in the same process grants nothing.

The registry indexes instances by id and get(id) is an exact lookup: a singleton is found by its kind, a collection member only by its own id, and a miss is undefined with no first-registered fallback. Ids are unique across the registry, so a duplicate id, a singleton under a custom id, or a kind mixing singleton and collection instances throws FlowIdentityConflictError at register. See the Engine API and server setup.

For voice, pass a voiceProvider (TTS + STT in one object); a per-flow voice.provider overrides it. See the Voice guide.

What this package does

• Action execution — Validates input, resolves sessions, runs block pipelines, emits items
• SSE streaming — Items stream live as blocks execute, with sequence-number cursors for resume. Resources declaring client: { live: true } emit their projected delta inline on each mutation so clients merge it without a refetch
• State persistence — in-memory, filesystem, SQLite, and Postgres store adapters. Version-checked writes for anything computed from current state; increments, appends and single-key writes are unchecked and apply to whatever the store holds
• Flow registry — Register multiple flows, and several instances of a collection flow, each addressed by its exact id; routes are derived automatically
• Instance ownership — Every session and request records the instance that created it (flowId, beside the definition's flowKind). Addressing a record through another instance, over HTTP, a transport, a worker or direct runAction, is refused with FlowInstanceBindingMismatchError before any effect (409 wrong-instance-session / wrong-instance-request on the action route). Session and request listings take an exact flowId filter and project flowId on every row; resolveRecordOwner / ownsRecord are the exported checks. pinRejectsCaller(pin, { userId, orgId }) (is a caller outside an instance's owner pin) and isOrgAttributed(record) / UnattributedOrgError (a record stored before organizations were required) are exported too, so an in-process caller such as fsdev run refuses before it writes exactly what the transport host refuses. Records with no owner recorded belong to the singleton of their kind; a collection kind with such history is migration-required until attributed (see Persistence)
• Token accounting — Every model result a generator or evaluator reports is counted in ctx.request.tokenUsage, and an action's tokenBudget is checked against it once the action's block returns (error fails the request, stop ends it incomplete, warn emits a warning item). Harness runs reported through reportHarnessSpend are kept apart from that count; a request-end hook receives both as RequestEnding.spend. See token budgets
• Error normalization — All errors become typed FlowError instances with codes, retry signals, and scope context
• Structured logging — Every action execution logs flow/action/block IDs, attempt numbers, timing, and summarized payloads

Inbound transports

Every entry point into the runtime — native HTTP, MCP servers, webhooks, scheduled actions, custom transports — implements the same InboundTransportAdapter contract. The envelope's flowKind is the exact instance address, carried unchanged; the host admits it against the owner of any session or request it names before writing anything. The built-in HTTP adapter is mounted automatically; createFlowApiRouter accepts an adapters option to mount additional transports onto the same host:

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

const router = createFlowApiRouter({
  registry,
  stores,
  adapters: [
    // createMcpTransportAdapter({ /* ... */ }),
    // createScheduledTransportAdapter({ /* ... */ }), // @flow-state-dev/scheduled
    // createWebhookTransportAdapter({ /* ... */ }),
  ],
});

Routes from every adapter merge into the returned { GET, POST, PATCH, DELETE } dispatcher. Path collisions among non-HTTP adapters throw TransportRouteCollisionError at construction time so dispatch is unambiguous at runtime. Every request carries a source field on its RequestRecord for provenance — http for the default adapter, set by each custom transport for its own.

An adapter can register a dedicated route that lives outside the canonical basePath — for example the MCP adapter's /mcp/:kind under dedicatedBasePath: true. A long-lived host that mounts the flow API under a prefix uses dispatchDedicatedRoute(router, req) to serve those: it matches ONLY the custom adapter routes and returns the matched Response, or null when none match — it never falls through to the canonical flow-API handler, so the flow API (list-flows, actions, sessions) stays reachable only under basePath. @flow-state-dev/node's serve() calls it in its not-found fallback; catch-all hosts that mount the router at basePath don't need it.

createWebhookTransportAdapter({ providers }) mounts POST /api/flows/:kind/webhooks/:provider and routes verified inbound webhooks (Stripe, GitHub, Slack Events, any signed POST) to the handler a flow bound in its webhooks config. The flow declares routing only; the host supplies signature verification and payload mechanics per provider at the mount, keeping secrets out of the flow definition. stripeWebhookVerifier, githubWebhookVerifier, slackWebhookVerifier, and createWebhookVerifier cover the common signature formats; each accepts a string secret or a () => string getter. See the webhook receivers reference.

See the inbound transports reference for the full contract reference and a walk-through of authoring a custom adapter.

A flow's concurrency policy is enforced once at the host dispatch seam — the in-process runner gates the run there (and, over a shared lease backend, the host takes an external run's place before enqueueing; see below), so every transport inherits the same behavior and adapters only map the outcome to their native response. When a reject policy drops a competing request, the handle host.dispatch returns refuses through accepted (and finished) with ConcurrencyRejectedError, carrying the contended key and the inFlightRequestId. The place is taken only after the caller's ownership of the session has been checked, on every backend, so a caller who does not own a session never holds its key; the HTTP adapter maps it to 409, fire-and-forget webhook/scheduled adapters to a benign skipped 200, MCP to a server-busy error. A queue policy that waits past its budget rejects the request's finished with ConcurrencyQueueTimeoutError (it surfaces through the request stream, not a synchronous status). Both errors are exported from this package. A hold run starts at once and marks its key busy; a defer run waits, with no time budget, until nothing on its key is running or waiting. At most 32 defers wait on a key per process (a defer handed to a queue counts until its job ends; see below); the next is refused with ConcurrencyDeferLimitError, a ConcurrencyRejectedError. A defer yields to newer hold runs for 30 seconds, then waits only for the runs it found. hold and defer are enforced across a queue whose adapter supplies a shared leaseBackend (see below). Over any other external runner, a run under either policy runs as allow. See the concurrency policies reference.

Concurrency across processes: WorkerAdapter.leaseBackend

By default the arbiter keeps each key's line in the running process (createInMemoryLeaseBackend()), so a policy serializes requests that run in that process, and work handed to an external queue runs unarbitrated. A queue adapter whose runs land in several processes can supply a backend they all share as leaseBackend on its WorkerAdapter; bullmqWorker supplies one on its Redis. The arbiter in every process then lines up on the same keys, and an external dispatch is arbitrated too: the host takes the run's place before enqueueing (once the caller's ownership of the session and request id is confirmed), and the job carries it as DispatchEnvelope.leasePlace.

A backend implements four calls and holds no policy:

| Call | Does | | --- | --- | | take({ key, requestId, ifEmpty? }) | Appends a place to the key's line, or with ifEmpty (the reject policy) claims only a free key and otherwise answers { heldBy }. Not idempotent | | isMyTurn(place) | True when the place is first on its key. "missing" when the line no longer has it (its lease ran out): the waiter takes a new place at the back | | giveBack(place) | Removes the place and wakes the next waiter. Idempotent | | renew(place) | Extends a place whose lease expires. Called every 2 seconds (every quarter of leaseMs when that is shorter) for as long as the process holds the place, until it gives it back or enqueues its job. Answers false when the place is gone |

A backend whose places expire also declares leaseMs, how long a place lives past its last renewal. A run holding its turn is then stopped when a renewal answers false, or when no renewal has landed for half of leaseMs, so it has ended before another process can take the key. The run's signal fires with ConcurrencyLeaseLostError and the request ends interrupted.

Any call may throw when the backend is unreachable; the dispatch is then refused rather than run unarbitrated. Over a supplied backend a reject refusal arrives through the handle's accepted and finished, not as a synchronous throw, and the built-in adapters map it the same way.

The worker owns a job's place: it waits for isMyTurn, runs, and gives the place back. planQueueWait({ key, waitedMs, attempt }) answers "not my turn yet" the way the engine does, with jittered backoff inside the 30-second queue budget, so a worker that waits by requeueing its job reaches the same timeout an in-process run would. holdLeasePlace(backend, place, onLost) renews the place while the job runs and reports it lost the way the arbiter does, and settleUnstartedRequest(stores, requestId, ending) ends a request whose run never started (a wait that timed out) with the same record the engine writes. A job whose leasePlace is null or absent runs as it always did.

hold and defer reach the worker as DispatchEnvelope.leaseTurn. A hold job carries { kind: "now" } with its place: it runs at once and gives the place back at the end, so the key reads busy until then. A defer job carries { kind: "when-free", key } and no place: the worker claims the key with take({ ifEmpty: true }) and runs under the place it gets.

While the key is held, planDeferWait({ waitedMs, attempt }) decides the job's next step. It returns one of two shapes:

• { kind: "claim-if-free", retryInMs, patienceLeftMs }: requeue the job for retryInMs (at most 2 seconds), then try take({ ifEmpty: true }) again.
• { kind: "line-up" }: the patience, DEFER_PATIENCE_MS (30 seconds), is spent. Take a place at the back with take(), then wait for its turn with planQueueWait({ ..., budgetMs: Infinity }), which never times out.

The dispatching process counts a dispatched defer against its 32-per-key cap until the job's finished settles, including while the job runs.

Authentication

Per-flow defineFlow({ authentication }) and a host-level resolvePrincipal (on createFlowState, or createFlowApiRouter for the lower-level surface) configure how the framework resolves the caller principal for every inbound transport. The framework owns the contract; the host owns credential verification.

import { defineFlow } from "@flow-state-dev/core";
import {
  createFlowApiRouter,
  createHmacVerifier,
  PrincipalResolutionError
} from "@flow-state-dev/engine";

const verifyStripe = createHmacVerifier({
  secret: process.env.STRIPE_WEBHOOK_SECRET!,
  format: "stripe"
});

const stripeFlow = defineFlow({
  kind: "stripe-webhook",
  authentication: {
    requireUser: false,
    defaultUserId: "system",
    resolvePrincipal: ({ rawBody, request }) => {
      const sig = request?.headers.get("stripe-signature") ?? null;
      if (rawBody === undefined || !verifyStripe(rawBody, sig)) {
        throw new PrincipalResolutionError("Invalid signature", { status: 401 });
      }
      return null; // defaultUserId fills in
    }
  },
  actions: { /* ... */ }
});

requireUser: false opts the flow out of user-scope identity at build time — defineFlow rejects user-scope state, clientData, and resource declarations on such flows. Bundled helpers createHmacVerifier (Stripe and GitHub-style signatures), createHs256JwtVerifier, createBearerSecretPrincipalResolver (constant-time bearer-token check for scheduled and webhook callers), and extractBearerToken cover the most common verification patterns; hosts plug in their own for anything else.

A configured resolver governs the flow's whole /api/flows surface, not only action calls: session CRUD, session state, resource content, request control, and the debug endpoints resolve a principal the same way, and additionally require that principal to own the session or request the URL addresses. Another user's session or request is answered exactly as an id nobody has used (a 404, or an empty 200 on a request stream resumed from a cursor), on every route that addresses an existing session or request. Creating a session is the exception: an id already in use answers 409, whoever holds it. Endpoints that span every flow (GET /sessions, GET /active-requests) resolve through the host-level fallback and scope their results to the caller. Reached without one, they return rows from flows with no resolver of their own to any caller. Either way, a row owned by a flow instance with its own resolver is listed only when that resolver identifies the caller as the row's owner (and, for an instance registered with an owner pin, { orgId, userId? }, only callers in that org, and that user when set). A caller a configured host-level fallback rejects gets 401 from both.

When no resolver is configured, a flow runs on the framework default that trusts a caller-supplied body.userId — unauthenticated, management surface included. isDefaultBodyUserIdPrincipalResolver(resolver) reports whether a resolver is that default, via a globally-registered brand rather than function identity (so it holds across duplicate package instances, e.g. a config that resolves its own copy of the engine). Tooling uses it to detect an unauthenticated flow before exposing it — @flow-state-dev/node's loopback-bind guard refuses a network bind when a served flow resolves to this default.

See the authentication reference for the full contract, resolution order, and requireUser: false semantics.

Multi-tenant isolation

Pass a tenant id on the x-tenant-id header (configurable via createFlowApiRouter({ tenantIdHeader })) and session storage namespaces by tenant automatically: two tenants sharing a session id get distinct session records, state, session-scoped resources, and request history. User and org scopes stay shared across tenants by design. Single-tenant apps that never send the header are unaffected — keys are unchanged and no migration runs. Read the value in a block via ctx.session.identity.tenantId. See the state and scopes guide.

Store configuration

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

// Default when no `stores` is passed: in-memory (dev/test only)
const router = createFlowApiRouter({ registry });

// Testing: in-memory (fast, no cleanup)
const router = createFlowApiRouter({ registry, stores: createInMemoryStores() });

Pass maxResponseBufferSize to cap how large a live SSE response the router buffers:

const router = createFlowApiRouter({
  registry,
  maxResponseBufferSize: 10_000,
});

Pass maxRecordedValueBytes to change the largest block output or tool result recorded in an item (default 256 KiB). Larger values are recorded as an omitted placeholder with their size and a 512-character preview; the run itself keeps the full value. A resumed request that would need such a value fails with RECORDED_VALUE_OMITTED.

createFilesystemStores wires a filesystem-backed trace store under {rootDir}/traces/ so trace events survive process restarts. Retention is controlled by traceStore.maxRequests, which defaults to 1000 when NODE_ENV=development and 50 otherwise — explicit values always win. See the trace channel reference for the full backend list and file layout.

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

Dispatched runs

Work a dispatcher starts runs in a session of its own on the same flow.

GET /api/flows/sessions?include=dispatch-runs lists a flow's sessions with those included. Rows are whole session records, and one a dispatcher started carries parentSessionId — the session it was started from — beside its topic and coordinate labels. Without the parameter the listing returns the sessions a person started; any other value answers 400. The owner, tenant and organization filters apply as they always do.

GET /api/flows/sessions/:sessionId/children is the provenance index for one session: which runs were started from it. The route is /children and one row is one dispatch run. Each row carries the run's id, the session it came from, topic and coordinate labels, timestamps, and a status of active (not finished) or a terminal outcome (completed, failed, aborted, incomplete). A run with no requests has no status. Those seven fields are the whole row — the route sends a named field set, not a session record.

topic and coordinate are stamped when the run's session is created, from the values its id was derived from: topic is the session key, coordinate is the entry the dispatch was addressed to (internal:<action> or task:<action>). Both are display only — nothing routes, authorizes or adopts on them — and both are optional, so guard with == null.

The route is session-addressed: the parent is loaded and ownership-checked before the handler runs, and the answer is scoped to the stored parent's owner, tenant, org and flow kind. limit accepts 1–100 (default 25) and offset 0–10000; anything outside returns 400. Use each row's id with the existing /sessions/:id/requests endpoint to read that run's history.

Those runs carry a metadata.dispatch bag on the request record: type and action for the entry, from naming the block and session that dispatched, key for the derived session key, and taskId naming the task-board row on a task dispatch. Read the bag at all only when the record's source is "internal" or "task" — metadata is caller-writable on an ordinary request, source is not. The runtime assembles the bag from values it derived itself, and it is a correlation rather than an authority: display taskId, don't key a settlement on it. key and taskId are optional, so guard with == null.

See Dispatched work for the full contract.

Following a whole session

GET /api/flows/sessions/:sessionId/stream is a server-sent-events stream of one session: each finished item from any request in it, including requests another tab, person or agent sent, and the session's unfinished runs. Like the snapshot, it shows only requests made under the session's owner and organization, so a request the server refused because its user or organization didn't match the session's never appears. It is authorized and refused exactly as the session snapshot (GET /api/flows/sessions/:sessionId/state) is, before anything is streamed. An unknown session answers 404, and one that needs migrating 409 (migration-required). In an app that authenticates, both routes can also answer 404 when the session is replaced by a new one under the same id while the request is being answered. The client's sessions.stream in @flow-state-dev/client reads it.

Query parameters, all optional:

• since: a server time (epoch ms) to start from, either the at of the last event you heard or, when the stream follows a snapshot, the snapshot's at. A snapshot's at is the time its read began, and everything saved before it is in the snapshot; when you read its items over several pages, use the first page's at. Pages are cut from the history as it stands at each request, so an item can fall between two pages if the history shifts during the read: start each later page one item early and read again if that item is not the one you last held, or if the page's sessionCreatedAt differs from the first page's. The stream starts a few seconds before since, so nothing is missed. Without it, the stream starts one minute back.
• session_created_at: the snapshot's sessionCreatedAt, the time the session it read was created. The stream then follows that session only: once the id holds a session created at any other time, a connection answers 404, or 403 if the app authenticates and another user or organization holds it.
• item_types: comma-separated item types, applied as on the snapshot route.

A connection reads the store when a change notice for the session wakes it, at most once a second. On a bus that crosses processes, the in-memory default included (it serves an app that runs as one process), that is its only read, so an open view of a session where nothing changes costs the store nothing. On a bus that doesn't cross (a Postgres bus with no tail pool, or one that has stopped crossing) it also reads about once a second, so a reply written by another process still arrives. Pings and the 15-minute close keep their own timers.

Each SSE frame's event is the event's type, and its data is the whole event as JSON. Every event carries stream: "session", sessionId, and at, a server time (epoch ms) to resume from. Reconnecting with any event's at as since misses nothing the connection hadn't already sent.

• session.item: { requestId, item }, one finished item from any request in the session. An item the session snapshot hides is never sent, and neither is one still being generated. Hold items by requestId and item.id together: two requests can save items with the same id. One request can also emit the same id more than once (a keyed component, say). Each emission is saved with its own ts and itemIndex and sent as its own session.item, even on a connection that sent an earlier copy, so keep the copy that sorts later. An exact repeat, with the same ts and itemIndex, can arrive twice, across a reconnect or because the stream starts a few seconds before since. Items from different requests arrive in no set order. compareItemOrder (from @flow-state-dev/client or @flow-state-dev/core/items) is the order the snapshot uses, ts, then itemIndex, then requestId, then id: sort merged items with it, and use it to tell which of two copies is later.
• session.runs: { runs }, every run under the session that hasn't finished, each { id, parentSessionId, createdAt, updatedAt, flowId?, topic?, coordinate? }. Sent when the stream opens and again whenever that set changes.
• ping: no other fields, sent when nothing else has gone out for 15 seconds.

The server reads the store about once a second per open stream, and each read costs what is running or recently changed, not the session's history. Every server reads the same store, so a line written on one instance reaches a view held by another. The server closes the connection after at most 15 minutes; reconnect with since to carry on, as the client's sessions.stream does. It also closes the connection when the session is deleted or replaced by a new session under the same id. The reconnect then answers 404 if the session is gone, or 403 if the app authenticates and another user or organization now holds the id. If the same owner created the session again, or the app doesn't authenticate, the reconnect answers 404 when it names session_created_at, and otherwise follows the new session. The client's sessions.stream stops for good, without an error, on a 401, 403, 404, 409 or 501.

A session deleted and created again under the same id, by the same owner in the same organization, treats the earlier session's requests and dispatch runs as its own: they show in its snapshot, its stream and its list of runs. Give a new session a new id.

A store adapter shows it serves the stream with createSessionStreamConformanceTests, from @flow-state-dev/engine/testing.

Runs as a row's owner

fireOwnerRunThrough(host, run) starts an internal entry as the user whose stored row it names. run is { flowKind, entry, input, owner: { userId, orgId, tenantId? }, rowKey }, where rowKey is the row's full key in a user-scoped collection that every flow shares (flowIsolation: false). The owner only says whose rows to look in, the same per-user, per-org storage the scheduled dispatch endpoint reads (resolveSharedUserCellKey). The run starts only when that user's rows hold rowKey, the row's orgId and tenantId are the owner's, and the row isn't enabled: false. Otherwise it answers { fired: false, refused } with no_row, other_org, other_tenant or disabled, and nothing runs. A run that starts answers { fired: true, requestId, finished }. Every run of one row starts in the same session.

routeServiceEvent(host, { flowKind, entry, input, principal }, { concurrency? }) runs a service flow's internal entry as the service's own principal, then fires each run its output lists in ownerRuns through fireOwnerRunThrough, at most concurrency (8 by default) at once. It answers { output, error?, ownerRuns, malformed }. Each owner run carries finished, refused, or failed (the host threw while starting it; the other runs still fire). malformed lists the entries it didn't fire: one naming another flow than the service's own, or an owner in another org. Workforce's subscriptions use it (SUBSCRIPTION_ROUTE_ENTRY). Reach the host from an inbound transport adapter's createBindings(host).

Change bus

A change notice says that something under a topic changed, never what. Whoever hears one reads again through the route it already reads with.

import type { ChangeBus, ChangeNotice, BusTopic } from "@flow-state-dev/engine";

// A topic: a scope, its id, the tenant ("" when single-tenant), optionally one resource.
const topic: BusTopic = { scope: "session", id: sessionId, tenantId: "" };
const stop = stores.bus!.subscribe(topic, (notice: ChangeNotice) => readAgain(notice.kind));

• ChangeNotice is { topic, kind, origin? }. Kinds: items (a request's items landed), request (a request was written), run (a session record was written: the topic's own session or a run under it), state (the session's, user's or org's state was written), resource (a resource a client can read was written, created or deleted; the topic names it in resource) and resync (the bus may have missed notices; read once). origin names the request that made the change, when one did.
• state and resource publish from any run, with or without a stream, and from the client's collection routes (with no origin). A resource publishes when it has a client config, or when its scope declares client.derived. Request and block-instance state publish nothing. A row only its owner may read (an ownerPrivate collection's) publishes on the bare scope topic, naming no path.
• A subscriber to a scope's topic hears every notice under it, narrowed to a resource or not, with the notice's own topic. A subscriber narrowed to a resource hears only that resource's.
• Notices are at most once: no replay, no acks, nothing stored.

Writing an adapter. A bus adapter declares "bus" in its capabilities and resolves { bus }, a ChangeBus:

| Member | Contract | |---|---| | publish(notice) | Called only after the change can be read. A failure never fails the write that caused it | | subscribe(topic, listener) | Returns the unsubscribe | | crossesProcesses | Whether a change made in any process that serves this app reaches this one's subscribers now. Only true turns the session stream's timed read and the client's fallback poll off | | onCrossingChange(listener) | Each change of crossesProcesses. A change to false puts waiting streams back on the timed read |

A bus that reconnects sends each subscriber one resync notice on its topic. createInMemoryBus() is the default, and the in-process fan-out an adapter can compose. It is the bus for an app that runs as one process, so it always says it crosses: with it, the session stream keeps no timed read and a client with an open socket stops its fallback poll. An app that writes its stores from several processes (several servers, a worker, or fsdev run / fsdev chat on the same SQLite file as a running server) needs a bus that crosses, such as the Postgres bus that Postgres stores fill by default. On the in-memory bus, the other processes' changes don't push to open pages.

A registry you build yourself. resolveStores, createFlowState, runAction, BullMQ's job processor and fsdev apply withChangeBus(stores), which gives the registry its bus (the one you set on stores.bus, else an in-memory one) and wraps its request and session stores so every write publishes. A host that writes to the stores past those entries calls it too, and writes through the registry it returns. It is idempotent: every entry gets the same wrapped stores.

Push delivery carries these notices to a browser. A host that serves the socket calls enablePushDelivery(router, { secret? }): the router then reports push: { transport: "websocket" } in GET /capabilities and answers POST /users/:userId/stream, which runs each topic the caller asks for through the read check its scope uses and signs a 30-second ticket for the ones it grants. The host hands each socket that opens on /users/:userId/stream to delivery.openSocket(userId, { send, close }) and moves text frames to the returned session's receive. The session accepts topics only from a ticket for that user, and closes the socket after an hour. It answers each accepted ticket with { type: "subscribed" } once the ticket's topics are heard, so a client reads after that point and misses nothing. After the first ticket, and each time the bus's crossesProcesses changes, it sends { type: "crossing", crosses }: whether a change from any process serving the app reaches this socket. A client keeps reading on its timer until it hears crosses: true. serve() in @flow-state-dev/node does all of this; secret defaults to a random key per router.

Store list options

SessionListOptions and RequestListOptions are part of the store contract. Adapters must implement all of these options:

• RequestListOptions.status accepts a RequestStatus or an array of them. An array matches set membership; an empty array matches nothing.
• RequestListOptions.sessionId accepts a session id or an array of them, matched the same way. The session stream uses it to ask about all of a session's child runs in one read instead of one read per run.
• RequestListOptions.orderBy accepts "none" alongside "startedAtMs" and "updatedAt". "none" returns the matching set unordered. An existence check needs that: its work must not grow with the set it selects on. "startedAtMs" orders by (startedAtMs, id) so an exact tie resolves deterministically.
• SessionListOptions.orderBy accepts "createdAt" or "updatedAt" (default). "createdAt" orders by (createdAt, id), both immutable, so a record written during a caller's walk cannot reorder its pages.
• Both types accept orgId, with the same present-vs-absent NULL-safe matching as tenantId: an absent key filters nothing, a present key (including an explicit undefined) exact-matches.
• SessionListOptions.state exact-matches top-level session-state fields: a record matches when each named field holds that string. Absent filters nothing. Filter in the query (the SQL adapters bind the field path), never after it.

Checking a session at create

Every path that creates a session (the create route, an action to a session id that does not exist yet, a webhook delivery, fsdev run, and a dispatch that creates its session) creates it from an initial state: the create body's state, fsdev run --seed-session, or the state a dispatcher() session target names. The flow's session stateSchema parses it. A flow that binds its sessions (a top-level .readonly() field or a createCheck) refuses a state the schema refuses; any other flow keeps it as sent, for its actions to validate when they run. session.serverOwned names fields a caller may not seed. Most flows need nothing more. The optional session.createCheck is for a rule that depends on who is creating the session or on what exists in the store: it receives the verified caller and the parsed state, and accepts or refuses the create. A turn on an existing session never calls it.

A dispatch creates a session through one of two targets. { key } derives a child session of the dispatching one from the key: it creates the child, from state when the target names one, or adopts the child already on that key. { id, state } creates a top-level session at that id when none exists, judged by the target flow's createCheck (create-refused when it refuses), and delivers into the session when it does exist. An { id } target without state never creates: an unknown id is refused as session-not-found.

A top-level stateSchema field declared .readonly() is set when the session is created and never changes: a run's write that changes it throws ReadonlySessionStateError and writes nothing, and fsdev run --seed-session refuses the same change. Only readonly fields are filterable on the list route (?state.<field>=<value>, with flowKind or flowId).

A host that creates sessions itself calls ensureSessionRecord(stores, key, request, build). It parses the request's state, runs the serverOwned refusal and the flow's createCheck, then writes the record you build with that state. If another caller created the same id first, it returns that caller's record. A refused create throws SessionCreateRefusedError, which carries the message and the HTTP status to answer with (and field, when one field was the reason). For a host that writes seeded state into a session that already exists, refuseServerOwnedState(flow, sessionId, state) throws the same error for a server-owned field, and refuseReadonlyStateChange(flowKind, sessionId, readonlyFields, createdWith, next) throws ReadonlySessionStateError for a changed readonly field.

Session retention policies

Long-running sessions accumulate items over time. Retention policies provide a safety net that bounds storage growth by evicting a session's old past turns when limits are exceeded: completed requests, and requests a dead host left interrupted, the same statuses the history view reads.

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

const flow = defineFlow({
  kind: "my-flow",
  session: {
    retention: {
      maxItems: 500,   // evict oldest requests when total items exceed 500
      maxAge: "24h",   // evict requests older than 24 hours
    },
  },
  actions: { /* ... */ },
});

Both constraints are optional and independent. When both are set, either condition triggers eviction. Eviction runs lazily after each completed request (no background process). The current request is never evicted. Nor is a request whose run has not finished yet, onFinished included, or that finished less than twice the live-tail liveness timeout ago (LIVE_TAIL_LIVENESS_MS, 30s by default, so 60s), or one still in the active-request registry. A completed run that dies before stamping finalizedAtMs is released by the stale-request sweeper, if it was heartbeating; with the sweeper disabled (staleSweepIntervalMs: 0), or heartbeats off (request.heartbeatIntervalMs: 0), its request is kept. A run that dies mid-turn is a different case: once the sweeper marks it interrupted it is evicted like any past turn, aged from interruptedAt (from its start, on a record without one), after the rollout-safety bound below has passed since it was marked. An interrupted request that will run again (a pending gate, or its last logged gate resolved and owed a re-drive) is never evicted, though its items still count toward maxItems. A request spared this way is evicted when the session's next request completes. A record from an engine version that never wrote finalizedAtMs is kept for the stale-request threshold or maxAge, whichever is larger, plus that window, so a rolling deploy cannot free the id of a run still finishing on an old instance. A custom RequestStore must persist the record's finalizedAtMs and heartbeatsUntilFinalized fields (the run writes the stamp through setFieldsIfStatus), and honor that method's expectedIncarnation fence.

Retention policies operate at request granularity — entire old request records are removed, not individual items. For items that should never be stored at all, use transient: true on block definitions.

The maxItems check counts items through RequestStore.countItems(requestId) rather than loading item payloads, so a retention sweep stays cheap on sessions with large logs. Custom RequestStore implementations must provide countItems; it returns what get(id) would surface as items.length.

Eviction calls RequestStore.delete(id), which removes everything the store keeps under that id: the record, its items, its stream events and its runOnce results. Request ids can be supplied by the caller, so a freed id may be taken by a later request, possibly from another user. Anything a custom store leaves behind under the id would be replayed on that request's stream. createRequestStoreConformanceTests checks the store's delete; createRequestRetentionConformanceTests (both from @flow-state-dev/engine/testing) runs the whole path through retention and the stream route against a StoreRegistry.

applyRetentionPolicy(stores, sessionId, currentRequestId, policy, now?, tenantId?, provider?) takes the host's durability provider as an optional last argument, to read those gates; runAction passes runtimeConfig.durabilityProvider. Without one, no request counts as waiting on a gate, which is right for a host that cannot suspend.

Supported duration formats: '30s', '5m', '2h', '7d', or a raw number in milliseconds.

Abort intent on RequestStore (adapter authors)

RequestStore carries two required members for cancellation. An out-of-tree adapter that leaves either one out fails the build with a missing-member error.

isAbortRequested(requestId: string): Promise<boolean>;

setFieldsIfStatus(
  id: string,
  fields: ConditionalRequestFields,
  allowedStatuses: readonly RequestStatus[],
  updatedAt: number,
  expectedIncarnation?: string
): Promise<ConditionalWriteResult>;

Both named types are exported from @flow-state-dev/engine. ConditionalRequestFields is a Partial of RequestRecord minus the fields that have their own write path: id, version, createdAt, incarnation, updatedAt, state, items, and the indexed access-path fields (flowKind, flowId, userId, sessionId, orgId, tenantId). abortRequested is in the set. status and result are in it only as a pair (ConditionalStatusTransition, also exported): a write carries both or neither. ConditionalWriteResult is { applied: boolean; status?: RequestStatus }.

isAbortRequested answers whether cancellation has been requested, without materializing the request. It runs on the heartbeat tick for the life of every request, so it must be O(1) in item count — reading the record and deserializing a growing item array turns a long run into quadratic work. Return false for an unknown request. Read the flag with === true; it is boolean | undefined.

setFieldsIfStatus applies fields only while the record's status is one of allowedStatuses, evaluating the predicate and the write as one atomic step. Three outcomes: { applied: true, status } when the predicate held; { applied: false, status } when a record exists outside the predicate; { applied: false, status: undefined } when no record exists.

A write that carries status moves the record to that status when the predicate holds, in the same atomic step, together with its result. A result of undefined removes any stored result. If your store keeps status in an indexed column, update the column in the same write, or a status-filtered list will not see the move. The returned status is still the status found before the move. The stale-request sweep uses this to mark a request interrupted only while it is still in_progress.

expectedIncarnation fences the write to one request. A request id can be reused once retention deletes its record, so the record at id when you write may not be the one the caller checked. When expectedIncarnation is given, compare it with the stored record's incarnation, resolved with resolveRequestIncarnation(record) (exported from @flow-state-dev/engine), and when they differ report the record as absent, { applied: false, status: undefined }, and write nothing. Do the comparison inside the same atomic step as the status check.

Resolve the stored side with the helper rather than reading record.incarnation directly. Records written before incarnations existed have none, and the helper gives them a stable value derived from createdAt. If your store has to state the rule in its own query language, the derived form is legacy_ followed by createdAt in decimal, used when incarnation is absent or null.

Don't compare createdAt instead. Two requests under one id can be created in the same millisecond, and a retry by the same owner rewrites createdAt on a request that is still running.

Do not implement it as a version CAS. Terminal transitions persist version unchanged, so a version-checked write still validates after a terminal commit and resurrects a dead record. The predicate has to read status.

setFieldsIfStatus is the only member that may write abortRequested. set must ignore the field in both directions: a full record handed to set cannot set the flag and cannot clear a stored one. The compiler will not enforce this, since the field is still on RequestRecord for reading. An adapter that honours it on set lets a full-record write built from a stale snapshot erase a cancellation.

How you store the flag is yours. The shipped adapters differ: in-memory and the SQL pair keep it on the record and force the stored value through on set; the filesystem adapter keeps a marker file beside the record, because its get() loads inline items and would break the O(1) bound. Whichever you pick, get() must still surface the flag on the returned record.

A set that leaves items off keeps the stored items. The runtime writes the whole request record when a block changes request state, and it leaves items off that record, because its copy is from when the run started. If your adapter keeps items on the record and set replaces the record, carry the stored items through whenever value.items is absent, or every item persisted since the run started disappears until the request settles. An adapter that keeps items out of set, as the SQLite and Postgres adapters do, already holds this. A record that does carry items may still replace them.

deleteIfStatus is the conditional form of delete, and is required too:

deleteIfStatus(
  id: string,
  allowedStatuses: readonly RequestStatus[]
): Promise<ConditionalWriteResult>;

It deletes what delete deletes (the record, its items, stream events and runOnce results), but only while the record's status is one of allowedStatuses. Check the status and delete in one atomic step, as setFieldsIfStatus does. Session retention deletes through it with the status it judged each record in, so a request that started running in between, such as an interrupted one the durability sweep re-drives, keeps everything under its id. The outcomes match setFieldsIfStatus: { applied: true, status } when deleted, { applied: false, status } when a record exists outside the predicate, and { applied: false, status: undefined } when no record exists. Wait for item and event writes already in flight for the id to finish, but don't drop them first the way delete does: when the predicate fails, they belong to a run that is still going.

The cross-store conformance suite (createRequestStoreConformanceTests, from @flow-state-dev/engine/testing) covers all of this.

Custom model resolution

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

const router = createFlowApiRouter({
  registry,
  modelResolver: createModelResolver(),
});

Custom logging

runAction and executeBlock emit structured logs by default — flow/action/block IDs, attempt numbers, summarized payloads, retries, and terminal errors.

await runAction({
  flow,
  actionName: "chat",
  input,
  userId: "user_123",
  stores,
  logger: {
    info: (msg, ctx) => appLogger.info({ ...ctx }, msg),
    warn: (msg, ctx) => appLogger.warn({ ...ctx }, msg),
    error: (msg, ctx) => appLogger.error({ ...ctx }, msg),
  },
});

Use summarizeForLog(value) for the same bounded payload summaries in custom log formatters.

Public API

Runtime:

• createExecutionContext — Build a block execution context
• runAction — Execute a flow action end-to-end. Also the sanctioned non-HTTP entry point (jobs, cron, queue consumers): pass an onItem callback to observe items live, and read requestId back off the result to correlate logs or attach a stream. Queue consumers re-running an action under the same requestId (retry attempts) pass startSequenceNumber — the last persisted sequence number — so the per-request event log stays strictly increasing across attempts
• executeBlock — Execute a single block with context

Stores:

• createInMemoryStores — Fast, ephemeral stores for testing
• createFilesystemStores — Persistent stores for local development (not for production load; use SQLite or Postgres in production)
• createInMemoryContentStore / createFilesystemContentStore — Content store adapters
• createInMemoryResourceStateStore / createFilesystemResourceStateStore — Resource state store adapters
• createInMemoryOrgStore / createFilesystemOrgStore — Org scope store adapters, implementing OrgStore. The filesystem adapter keeps org records under {rootDir}/projects/.
• Session, user, and request scope store factories, and CAS/state ops
• createInMemoryBus / withChangeBus and the ChangeBus, ChangeNotice, BusTopic types — the change bus (see Change bus)

Streaming:

• createResponseEmitter — Create an SSE emitter for a request
• ResponseEmitter.emitContentAudioDelta(itemId, contentIndex, chunk) — Emit a chunk of streamed TTS audio. Non-replayable; live-only.
• encodeStreamEvent / serializeSSEFrame — Low-level SSE encoding
• replayRequestEvents — Replay events from a sequence cursor

Request abort:

• abortRequest(requestId) — Signal an in-progress request to stop via AbortController
• hasActiveAbortController(requestId) — Check if a request can be aborted
• Abort endpoint: POST /api/flows/:flowKind/requests/:requestId/abort — returns 204 when the running process was signalled directly, 202 when the cancellation was recorded for another process to pick up, 404 when no request exists under that id, 409 when the request is no longer in_progress. See Connection Resilience for the cross-process path and what a 202 does and doesn't promise
• Aborted requests receive status: "aborted" with an abortedAt timestamp. The SSE stream emits request.aborted and closes.
• Background .sideChain() tasks survive client disconnect and only abort on explicit cancellation (POST /abort or session.abortRequest()). See the sequencer side-chains reference for the two-signal cancellation contract.

Registry/routes:

• createFlowRegistry — Register flow instances
• createFlowApiRouter — Generate HTTP route handlers from a registry
• parseFlowRoute — Parse an incoming method and decoded path segments
• decodePathSegments — Turn a raw, still-encoded URL path into those decoded segments

Cross-flow schema validation:

FlowRegistry.register validates each non-isolated flow's user.stateSchema, org.stateSchema, and user/org resource schemas against every other registered flow. Incompatible declarations throw CrossFlowSchemaConflictError at registration time — no silent data loss when a second flow's write would overwrite the first flow's keys. Flows that opt into isolation (isolateUserState: true or isolateOrgState: true on defineFlow) are namespaced by the flow instance id in storage and skip the registry check. A singleton's instance id is its kind, so its keys are unchanged; two registered copies of a collection definition each get their own cell. The same coordinate keys a resource declaring flowIsolation: true, for both its state and its content.

The exported scope-key helpers take an instance-bearing shape, and the user key takes the organization:

resolveUserStorageKey(userId, orgId, { id: flow.id, isolateUserState: false }); // "<userId>:~or