@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-5.4-mini" },
  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.
• 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(seat);      // one instance at a time
flowstate.unregister(seat.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.

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" },
    }),
});

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;

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. 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)
• 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 dispatcher gates the run there, 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, host.dispatch throws ConcurrencyRejectedError synchronously (carrying the contended key and the inFlightRequestId); 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. See the concurrency policies reference.

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 (403 otherwise). 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 serve only the flows that configure no resolver of their own.

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,
});

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 hand-off. 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.

Store list options

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

• RequestListOptions.status accepts a RequestStatus or an array of them. An array matches set membership; an empty array matches nothing.
• 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.

Session retention policies

Long-running sessions accumulate items over time. Retention policies provide a safety net that bounds storage growth by evicting old completed request records when limits are exceeded.

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.

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.

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
): 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, updatedAt, state, items, status, and the indexed access-path fields (flowKind, userId, sessionId, orgId, tenantId). abortRequested is in the set. 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.

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.

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

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 incoming request paths

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 therefore take an instance-bearing shape:

// before
resolveUserStorageKey(userId, { kind: flow.kind, isolateUserState: true });
// now
resolveUserStorageKey(userId, { id: flow.id, isolateUserState: true });

resolveOrgStorageKey, resolveResourceScopeId and resourceScopeIds change the same way; their return type is unchanged. A FlowInstance satisfies the input as-is. Attributing an existing collection deployment's stored cells to the copy that owns them is one offline procedure — see Persistence; there is no runtime fallback to the old kind-keyed cell.

See Flow Isolation and the state and scopes reference for the full model.

Execution backend (worker adapters):

• WorkerAdapter / WorkerHandle / WorkerMode — Contract for the worker option on createFlowState. An adapter (e.g. bullmqWorker from @flow-state-dev/bullmq) provides the dispatch side and/or the processing side; createFlowState hands both the same resolved { registry, stores, runtimeConfig } so the worker can never run against different stores than the router. mode picks the deployment shape: "colocated" (default), "dispatch-only" (web container), "worker-only" (worker container — call ready() to start consuming). dispose() drains the worker before closing stores

Dispatcher (pluggable execution backend):

• FlowDispatcher — Interface controlling where flow actions execute. Default: in-process. Implementations route execution to external workers (e.g., BullMQ)
• DispatchEnvelope — Serializable subset of InboundRequestEnvelope carried over the queue
• FlowDispatchHandle — Handle returned by dispatch(): requestId, finished promise, abort() hook
• createInProcessDispatcher — Default dispatcher that calls runAction in the current process
• StreamBridge / StreamPublisher / StreamSubscriber — Bridges live SSE events between a remote worker and the web process. The worker writes events to the bridge; the web process reads them and forwards to SSE
• StreamEvent — Single event published through the bridge, matching the SSE event shape
• Pass dispatcher to createFlowState or createFlowApiRouter to route all action dispatches through a projected queue. Most deployments should prefer the worker option — the adapter wires the dispatcher and the worker together; dispatcher is the low-level escape hatch (mutually exclusive with worker)

Errors:

• FlowError and canonical subclasses
• normalizeError — Wrap any thrown value into a typed FlowError

ContentStore

StoreRegistry includes a required content: ContentStore field that separates resource content persistence from scope record persistence. Both createInMemoryStores() and createFilesystemStores() include a default ContentStore. The filesystem ContentStore writes each resource as a nested .md file (a key concepts/overview becomes concepts/overview.md), so the store root is a browsable file tree; a directory written in the older flat layout is refused with a clear error rather than silently misread.

interface ContentStore {
  get(scopeType, scopeId, resourceKey): Promise<string | undefined>;
  set(scopeType, scopeId, resourceKey, content): Promise<void>;
  delete(scopeType, scopeId, resourceKey): Promise<void>;
  getAll(scopeType, scopeId): Promise<Record<string, string>>;
  getByPrefix(scopeType, scopeId, keyPrefix): Promise<Record<string, string>>;
  deleteAll(scopeType, scopeId): Promise<void>;
}

Per-request loading is scoped to the resources a flow declares: the execution context reads fixed resources with get and collections with getByPrefix (an empty prefix loads every key in the scope), rather than getAll. getAll remains for the state endpoint's full-scope view.

That scoped load runs in three waves. createExecutionContext fires Wave 1 (flow-level resources, at context creation) and Wave 2 (the dispatched action's declared resources, in one parallel burst — a context is bound to exactly one action, so this lives in the context rather than runAction). Wave 3 fires in the block runtime's run: a block's prefetchMode: 'lazy' single resources load when that block dispatches, and lazy collections defer further to a per-access on-demand accessor. A per-scope cache plus a single-flight in-flight map dedupe loads across all three waves and concurrent block dispatch. See the resources reference for the full model.

For custom store registries, provide a ContentStore implementation. createInMemoryContentStore() is the simplest option:

import {
  createInMemoryContentStore,
  createInMemoryResourceStateStore,
  createInMemoryCheckpointStore
} from "@flow-state-dev/engine";

const stores: StoreRegistry = {
  session: mySessionStore,
  request: myRequestStore,
  user: myUserStore,
  org: myOrgStore,
  activeRequests: myActiveRequestRegistry,
  content: createInMemoryContentStore(),
  resourceState: createInMemoryResourceStateStore(),
  checkpoints: createInMemoryCheckpointStore(),
};

Database adapters can implement ContentStore to route content to blob storage, S3, or a separate table while keeping scope metadata in the primary store.

Migrating from inline resourceContent: Earlier versions stored content inline on SessionRecord/UserRecord/OrgRecord as a resourceContent: Record<string, string> field. That field has been removed. Operators with content already persisted inline must copy it into ContentStore before upgrading — for each scope record, walk its old resourceContent map and call stores.content.set(scopeType, scopeId, key, value) per entry. After the migration the field is silently dropped on the next record write.

ResourceStateStore

StoreRegistry includes a required resourceState: ResourceStateStore field that separates resource state persistence from scope record persistence. It holds the structured JsonObject each resource carries (single resources and collection instances alike), keyed by (scopeType, scopeId, resourceKey). Both createInMemoryStores() and createFilesystemStores() include a default ResourceStateStore. The filesystem adapter writes each resource as a nested .json file mirroring the content store's layout (same nested-tree upgrade and legacy-layout guard).

It shares ContentStore's addressing but not its concurrency model. ContentStore is last-write-wins, which is right for a document body nothing merges against a prior read. Resource state is read-modify-written by concurrent workers, so the contract is compare-and-swap: every write takes an expectedVersion and returns a SetResult saying whether it actually landed.

That guarantee reaches flow-authored mutations, not only callers holding the store directly. The runtime drives every resource write through a CAS retry driver at the registry's read/mutate seam, passing the version the execution context observed: a conflict re-runs the op's mutator against the value that won and retries, so two contexts patching different fields of one resource both land.

Conflicts report what actually happened rather than collapsing into one error:

| Situation | Result | |---|---| | Key never persisted, write asks for no change | Verified no-op — not an error | | Held a live version, row is now a tombstone | ResourceDeletedError, terminal | | A create lost its race | ResourceAlreadyExistsError, terminal. getOrCreate and upsert absorb it and complete as a read / patch | | A delete's version check failed against a live row | ConcurrentModificationError — nothing was deleted | | Retry budget exhausted | ConcurrentModificationError |

The driver is deliberately separate from the one the four scope stores use (runWithCAS), which treats every conflict as retryable, suppresses a no-op before checking any version, and has no cancellation. The full policy table lives in the stores/resource-cas.ts module header. Resource writes honour the request's background abort signal, so a user-requested abort stops them — while a client disconnect does not, since background .sideChain() tasks keep running and their writes must land.

A resource stateSchema must parse its own output unchanged

A single resource's state is parsed on the way out as well as on the way in. The read path normalizes the stored row, your updater builds its next value on top of that, and the write parses the result. So anything the schema rewrites runs twice per read-modify-write cycle, once on each parse.

Collection instances are read back as stored. The read path does not normalize them, so a rewrite runs on an instance once, on the write. The count differs; the fact that the rewrite recurs does not.

That is fine when the rewrite settles. Filling a .default(), stripping an undeclared key, normalizing a retired enum value — all land on the same value the second time, so the row converges and then holds. It is also how a row written before its schema gained a field picks that field up. A whole-row .catch() fallback — including one sitting under .nullable(), .default(), or .readonly() — is different on the write path: those catch wrappers are peeled, and the candidate must satisfy the wrapped inner schema before fallback-normalized output can be stored. Field-level .catch() remains ordinary Zod normalization.

A .transform() that returns something different on each pass is the case that does not settle. Under z.object({ n: z.number().transform((v) => v + 1) }) the stored n climbs on every write even when the caller never touches it, and on a single resource the value you read back still looks plausible because the same shift re-applies on read.

A write through such a schema is refused:

Resource "counter" write failed stateSchema validation at "n": the schema does not
parse its own output back to the same value, so every write would move the stored
state. Make the transform idempotent — parsing an already-parsed value must yield
that same value.

A schema that collapses its own output — one whose second parse returns null or another non-object rather than a different object — is refused the same way, and says so instead of naming a field that did not move.

Every resource write runs the check, because they all share one parse path: setState / patchState / updateState / incState / pushState / getOrPatchState, the same ops on collection instances, collection.create() and upsert, and the client create route POST /sessions/:id/resources/:ref. That route carries no initial state, so it seeds the row from the schema's parse of {}, and a schema that cannot produce a valid, settled object from {} answers 400 rather than creating a row every later write would reject. A required field with no .default() is the usual cause; give it one.

A stored row that does not satisfy the check is read normally either way, but only one of the two kinds heals on its own:

| The row's schema | What happens to the row | |---|---| | Parse settles | Read normally, and its next successful write stores a settled value | | Parse does not settle | There is no next successful write — every mutation through that schema is refused, so nothing changes until the schema is fixed |

For the second kind, make the parse idempotent first; the next write after that settles the row. A value that already drifted keeps what it drifted to until a write corrects it. If you need a derived value, compute it where you read the state rather than inside the state schema.

// branded — see the note under the table below
type VersionedResourceState = { state: JsonObject; version: number };

interface ResourceStateStore {
  get(scopeType, scopeId, resourceKey): Promise<VersionedResourceState | undefined>;
  set(scopeType, scopeId, resourceKey, state, expectedVersion): Promise<SetResult<JsonObject>>;
  delete(scopeType, scopeId, resourceKey, expectedVersion): Promise<SetResult<JsonObject>>;
  getAll(scopeType, scopeId): Promise<Record<string, VersionedResourceState>>;
  getByPrefix(scopeType, scopeId, keyPrefix): Promise<Record<string, VersionedResourceState>>;
  deleteAll(scopeType, scopeId): Promise<void>;
  purgeTombstones(scopeType, scopeId): Promise<void>;
}

expectedVersion is a non-negative integer, "any" to write unconditionally, or "absent". Two meanings differ from the scope stores that share these types, and both are deliberate:

• 0 means "no live row" — create-if-absent. A tombstoned key satisfies it just as a never-existed one does.
• Some conflicts are terminal. A conflict against a deleted resource must not be retried into a resurrection, and a losing create must not be retried into an overwrite.

A number outside that domain — negative, fractional, NaN, Infinity — throws. TypeScript's number | "any" | "absent" admits it, but the contract has no meaning for it, so it is a mistake at the call site rather than a lost race. It is not reported as a conflict: that would name a concurrency outcome the store never observed, and send the caller into a retry loop that can never converge.

"absent" is the stricter of two create expectations on set, and throws on delete. It means what it means in the scope stores — "no record exists" — and here a tombstone is a record, so it refuses one where 0 admits it. The pair is what lets the store tell a key nothing ever wrote from one that was written and deleted:

| Write | Admitted when | Used by | |---|---|---| | set(…, 0) | No live row — never existed, or tombstoned | Explicit creation, including recreating a deleted resource | | set(…, "absent") | No row at all — a tombstone conflicts | A read-modify-write that started out holding no version, so it cannot undo a delete it never saw |

delete(…, "absent") still throws. 0 already covers "no live row, so the requested terminal state holds", and "delete only if absent" asks nothing on top of that — the same call assertDeltaExpectedVersion makes on the scope side's delta verbs. Keeping the refusal to that one verb is what stops the word acquiring a second, verb-dependent meaning.

Versions, deletes and retention

| Rule | Behaviour | |---|---| | Reads | get / getAll / getByPrefix return live rows only. A deleted key reads exactly like an absent one | | Version | First create writes 1; each committed write adds one; never reused. A recreate continues from the tombstone's version | | delete | Takes a version like every other write, retains it, and drops the payload. A delete chosen from a stale snapshot conflicts rather than tombstoning a newer generation | | delete(…, "any") | Never reports a conflict. Finding no live row is the whole answer to a blind delete, so it succeeds at version: 0 even if a concurrent recreate makes the key live again a moment later. A positive expectedVersion in the same race still conflicts — it asserted something that did not hold | | deleteAll | Bulk-marks every live key in the scope. A scope operation, so it carries no expected version | | purgeTombstones | Removes the scope's tombstoned rows outright and touches no live one. The only operation that reclaims a tombstone | | Retention | The store never reclaims a tombstone on its own — no sweep, no TTL, in any scope. Only an explicit purgeTombstones removes one | | Legacy rows | A row written before versioning reads as live at version 1 — never as absent |

Retention is the guarantee, not an oversight: because a tombstone keeps its version, an observer holding a pre-delete version can never match the row that replaces it within the same incarnation of a scope id. A tombstone that is never aged out is always sound. It costs one row per deleted key.

purgeTombstones is the deliberate exception, and it exists because scope ids are reusable. Session ids are caller-supplied, so chat-42 can be deleted and used again — and SessionStore.delete keeps no tombstone of its own, so the id really is free. The resource-state tombstones of the dead session would otherwise outlive it and refuse every "absent" write the new one makes, leaving each static resource permanently unwritable (a static ref has no create-if-absent verb to fall back on, unlike a collection instance). So the engine reclaims them at the moment they stop applying: when a new session record is born under that id, not when the old one died. While the session is merely gone, the tombstones still do their job.

It removes tombstones only. A live row is data — state can legitimately be written under a scope id before that scope's record exists — and a blanket purge would delete it. The cost is that a reclaimed key's version restarts at 1, so a straggler from the previous incarnation holding version N can match a row in the new one. That window opens only after a deliberate re-create under a reused id, and closing it properly needs a scope generation rather than a per-key predicate.

The engine runs it immediately before creating the session record, not after. There is no transaction across the two stores, and creating first would leave a committed record over intact tombstones whenever the reclamation failed — with nothing to retry it, since a second create answers 409 and an action-driven create adopts the record instead. Reclaiming first commits nothing until it has succeeded, and running it twice is a no-op.

Writing an adapter: purgeTombstones is required. It is one statement in SQL (DELETE … WHERE scope_type = ? AND scope_id = ? AND lifecycle = 'deleted'), and the shared conformance suite covers it. Adding a required method to a published store interface breaks every implementer — including test doubles, which are implementers too.

toBareState / toBareStates are exported for readers that only want the stored value and not the version beside it. VersionedResourceState is branded, so it is not assignable to JsonObject and a missing unwrap fails to compile rather than silently handing the wrong shape downstream. The brand is a phantom optional property that never exists at runtime — adapters still construct a versioned read as a plain object literal. It does not defend against an explicit as cast; that stays the caller's assertion to make.

Both are generic in the state shape, so a caller that knows what it stored names it in the call instead of casting the result:

const rows = toBareStates<InboxRecord>(await stores.resourceState.getAll("user", userId));
const one = toBareState<InboxRecord>(await stores.resourceState.get("user", userId, key));
const bare = toBareStates(rows); // no type argument → Record<string, JsonObject>

The type argument is asserted, not validated — nothing checks the stored row against it at runtime, exactly as the cast it replaces did not. What it buys is that the assertion is named and greppable at the call site rather than hidden in an as unknown as. The JsonObject bound keeps it honest: it rejects any shape the store could not have held (a Date field, say) and rejects VersionedResourceState itself, so the projection cannot be used to launder the versioned shape back in.

A delete is idempotent at the terminal state: deleting a key that is already absent or already tombstoned succeeds and reports the retained version, and that holds under a race — two concurrent deletes of one live key both report success. A conflict is reserved for a version mismatch against a row that is still live.

Per-adapter guarantee

Real compare-and-swap on in-memory, SQLite and Postgres. The filesystem adapter compares under a per-key mutex held on the store instance: it closes the race between two execution contexts that share that instance, and does not coordinate two instances over one directory, whether they sit in one Node process or two.

The collection-item HTTP routes use it

The two client-facing write routes are the callers that pass a real expectedVersion today. Both follow one rule: settle the state key first, then touch content, so a request that loses a race never reaches ContentStore.

• POST /sessions/:id/resources/:ref inserts the state row at expectedVersion: 0 (create-if-absent) and writes content only after that commits. Two clients creating the same topic get one 201 and one 409, and the stored body belongs to the client that won; on the filesystem adapter that holds only among writes through one store instance, per the per-adapter guarantee above. The conflict is terminal — a losing create is never retried into an overwrite.
• DELETE /sessions/:id/resources/:ref/:topic reads the row's version, deletes state conditionally on it, and deletes content only after that commits. If the row moves between that read and the delete, the request returns 409 and leaves the item intact in both stores. Deleting an absent topic is still an idempotent 200. The version is one the route observes, not one the client supplied, so the window it closes is the route's own — a DELETE from an out-of-date client view still reads the live row and removes it.

Because each route writes two stores, some windows stay open: an item can exist with no content row after a create whose content write failed — the state row commits first, so a failed POST is not a no-op, and the item is live and listable but reads back as content: null rather than "" — a create's body can be orphaned by an overlapping delete, an acknowledged PATCH can be overwritten by an in-flight create, and a recreation can lose its content to a delete already in flight. Only the first surfaces as an error; the others are silent. A collection whose content comes from contentTemplate / contentTemplateRef loses only the repair half of the first — it renders from state and never reads the content row, so it stays readable — but the create route does not guard its content write on the template config, so the half-commit is identical there. These are accepted — content is deliberately unversioned, so no state predicate fences a write to it, and closing them is cross-record atomicity, which this store does not provide. The full contract, with the reasoning and the residual table, is in docs/architecture/resources-and-client-data.md; client-facing guidance is in the resources / client access docs.

CheckpointStore

StoreRegistry includes a required checkpoints: CheckpointStore field for durable sequencer checkpoints. Sequencers default to durable: true and overwrite a single record per (requestId, blockInstanceId) at every step boundary; the future durable execution runtime reads latest(...) to resume after an interruption.

interface CheckpointStore {
  write(checkpoint: SequencerCheckpoint): Promise<void>;
  latest(requestId: string, blockInstanceId: string): Promise<SequencerCheckpoint | null>;
  delete(requestId: string, blockInstanceId: string): Promise<void>;
  deleteForRequest(requestId: string): Promise<void>;
}

deleteForRequest removes every checkpoint for a request in one call (all blockInstanceIds) — used by the retention sweeper to reclaim checkpoints of a crashed run whose per-instance terminal deletes never fired.

Memory, filesystem, SQLite, and Postgres adapters all ship with first-class implementations. Per-sequencer storage is constant regardless of step count.

By default the final checkpoint is retained after terminal completion (success / error / abort) for post-mortem inspection. Set flow.request.cleanupCheckpointsOnTerminal: true on a flow to make terminal frames trigger an immediate delete().

DurabilityProvider

DurabilityProvider coordinates checkpoint-based crash recovery and HITL (human-in-the-loop) suspend/resume. Wire it onto RuntimeConfig.durabilityProvider to enable ctx.suspend() in durable actions.

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

const provider = createCheckpointDurabilityProvider({
  checkpoints: stores.checkpoints,
  suspensions: stores.suspensions,
  leases: stores.leases
});

// Pass to runtime config
{ durabilityProvider: provider }

The interface methods are saveCheckpoint, loadCheckpoint, suspend, loadSuspension, listSuspended, acquireLease, releaseLease, cleanup, plus the retention seams cleanupCheckpoints (delegates to CheckpointStore.deleteForRequest) and pruneSuspensions (delegates to SuspensionStore.pruneTerminalBefore). createCheckpointDurabilityProvider delegates each to the matching store from StoreRegistry.

SuspensionStore and LeaseStore ship with in-memory, filesystem, SQLite, and Postgres adapters. See the Durable Execution guide for usage patterns.

A suspension inside a router's chosen branch resumes the same branch: the recorded router_decision is validated against the re-run selector before dispatch (a mismatch fails with RouteUnavailableError), and completed work inside the branch replays from the durable log instead of re-executing.

A generator tool can also suspend mid-loop (ctx.suspend() for tool-call approval). On resume the conversation is rebuilt from the durable item log rather than re-calling the model: prior model turns and completed sibling tools replay from their recorded items, and only the gated tool re-enters to produce its real result. See the Generator and router suspend/resume reference for the contract and limits.

Durability retention

Durability records accumulate on long-lived hosts: a completed run's checkpoints are dead weight, a resolved suspension is only worth keeping for a window, and a crashed run leaves records that cleanup() never fires for. createDurabilitySweeper is an opt-in periodic job that reclaims them, modeled on the stale-request sweeper (setInterval + unref, inFlight guard, idempotent dispose, no-op handle when disabled).

Configure it via RuntimeConfig.durabilityRetention (forwarded by createFlowState and createFlowApiRouter). The sweeper is built only when both a durabilityProvider and a durabilityRetention policy are present.

createFlowState({
  // ...
  durabilityProvider: createCheckpointDurabilityProvider(stores),
  durabilityRetention: {
    sweepIntervalMs: 600_000,                // sweep cadence; 0 disables. Default 10min.
    checkpointMaxAgeMs: 86_400_000,          // terminal-run checkpoint backstop. Default 24h.
    suspensionTerminalMaxAgeMs: 604_800_000, // resolved-suspension window. Default 7d.
    orphanCheckpointThresholdMs: 86_400_000, // abandoned-interrupted threshold. Default 24h.
    batchLimit: 1000,                        // max deletes per store per tick. Default 1000.
  },
});

Each tick takes a single-holder sentinel lease (co-located hosts serialize), enforces suspension expiry (pending past expiresAt → expired), prunes resolved suspensions and expired leases past their windows, and prunes orphaned checkpoints. Checkpoints of in_progress or suspended requests are never age-pruned — they are the resume points an active or paused run needs.

Connection resilience

The server runs three coordinated mechanisms so a dropped SSE connection doesn't leave a request running forever with no way to recover:

• Wire-level SSE heartbeat. Every live and GET-attach SSE response emits : ping\n\n comment frames at a configurable cadence. Keeps NAT/proxy idle timeouts from closing the connection and gives clients a robust inactivity signal during long pauses (e.g. an LLM thinking).
• Stale-request sweeper. A periodic in-process job that reads the active request registry and, for entries whose executor heartbeat has stopped, marks the persisted request record interrupted so session locks release.
• Read-only status endpoint. GET /api/flows/:flowKind/requests/:requestId/status returns a RequestStatusSnapshot callable when no SSE stream is attached — used by clients to confirm authoritative server state after the watchdog trips.

createFlowApiRouter({
  registry,
  stores,
  // SSE wire heartbeat — applied to every live and GET-attach stream when
  // the per-flow `request.sseHeartbeatMs` is unset. Default 15_000 ms.
  defaultSseHeartbeatMs: 15_000,
  // Internal sweeper cadence (0 disables). Default 30_000 ms.
  staleSweepIntervalMs: 30_000,
  // Heartbeat-age threshold the sweeper uses to mark a request `interrupted`.
  // Should be ≥ 2× the executor's registry heartbeat. Default 60_000 ms.
  staleSweepThresholdMs: 60_000,
  // How long a request queued with a projected dispatcher may wait, unclaimed,
  // before a sweep treats it as lost. Default 600_000 ms (10 minutes).
  queuedGraceMs: 600_000,
  // Sources from your own inbound transports that `retry` / `continue` /
  // `resume` may re-enter. The built-ins (`http`, `mcp`, `scheduled`)
  // are always admitted; every other source is refused with a not-found unless
  // named here. `webhook`, `task` and `internal`
  // are never openable — naming one throws at construction.
  publicReentrySources: ["echo"]
});

Per-flow override (wins over the host-level default):

defineFlow({
  kind: "chat",
  request: { sseHeartbeatMs: 10_000 },
  actions: { /* ... */ }
});

Clients consume the wire heartbeat through useSession's watchdog: it surfaces session.isStuck when the stream goes silent past stuckThresholdMs (default 30s) and exposes session.dismissRequest() to clear the request without a live connection. See Connection Resilience for full details.

Debug endpoints

The server exposes a read-only debug surface at /api/flows/sessions/:id/debug/resources and /api/flows/sessions/:id/debug/resources/:ref. Each response carries the full server-side state for the matching storage keys alongside the projected client view, so a debugger can show you exactly what client.data is dropping. There are no write paths here; the endpoint cannot mutate state. A response lists one entry per resource. Each entry reports the resource's writable and llmWritable settings where the config declares them. A setting the config does not declare is left out of the entry rather than reported as false. A projected collection has no writable field. Its debug entry still reports writable: false, unlike a resource that simply omits the setting.

The endpoint is off by default. Opt in with debugEndpointsEnabled: true on createFlowApiRouter, or set FSDEV_DEBUG_ENDPOINTS=1 in the environment. By default the route accepts only loopback origins; widen with debugAllowedOrigins for non-loop