ElementsFlow

Overview

One Flow species for HTTP, Signal, Clock, CDC, and call-only work — contracts, fx, and options in one place.

Flow is the element for behavior. An HTTP endpoint, a Signal worker, a named Clock tick, a SQL change handler, and a multi-step checkout are the same shape: on(trigger, flow). Only the trigger changes.

For developers writing backend work on okengine — put the invoke contract on the exposure (http.* / call / mcp.tool), keep do on fx.

The one rule

All world access goes through fx. A direct fetch, Date.now(), or node: import inside do is a defect. Effects are inferred from what the Flow touches through fx — that inference powers the Manifest, Console, cache, and durability.

on(Trigger) → Effects

one species
  1. 01Trigger
    any trigger → one flow
    httpclocksignalcdcmcp
    on(http.post("/orders"), createOrder)

    How work starts. Only this piece changes between an endpoint, a job, a consumer, and a row hook.

  2. 02Contracts
    exposure invoke contract
    inouterrorsexposure
    http.post({ in: z.object({ sku: z.string() }) })

    Invoke contracts live on the exposure (HTTP, call, MCP). Manifest projects in/out/errors onto flows.*.

  3. 03do + fx
    no side-channel I/O
    dofxsingle door
    await fx.store(db).insert(orders).values(input)

    The body. Every read, write, emit, secret, and model call goes through fx — side-channel I/O is a defect.

  4. 04Effects
    auto-derived manifest
    inferredmanifest diff
    writes: ["sql:orders"] · emits: ["orderPlaced"]

    Recorded from fx touches. Cache keys, capability tokens, live queries, and Manifest Diff fall out — no hand annotations.

Smallest Example

Define a Flow

src/flows/main/health.ts
import { on, flow, http } from "okengine";

export const health = on(
  http.get().public(),
  flow({
    do: () => ({ ok: true }),
  }),
);

Call it

curl -X GET http://localhost:6530/health -H "accept: application/json"

Response:

{
  "data": { "ok": true },
  "error": null
}

Omit path and name

Tree default: http.get() + flow({ do }) — no path or name strings. Pass either only for control.

Call-only Flows

Use call("payments.charge", { in, out, do, … }) for internal callees — same species as flow, with the invoke contract on the bag. Nothing outside your code can start it unless you also bind a trigger. Other Flows invoke it with fx.call(flowRef, input).

Progressive Patterns

Same on + flow + do from a ping to a typed failure to a private callee:

Return a value. HTTP wraps it as { data, error: null }:

src/flows/main/ping.ts
import { on, flow, http } from "okengine";

export const ping = on(
  http.get().public(),
  flow({
    do: () => ({ status: "ok" }),
  }),
);

Trigger Reference

flow, do, and fx never change. Bind a different trigger:

Five triggers, one species

on(trigger, flow)

one Flow

orders.create

Binding

on(http.post("/orders"), createOrder)
in: { sku: "desk-mat", qty: 2 }

HTTP requests validate JSON body, query params, and headers directly into in.

Contracts, do, effects — identical shape. Only the trigger changed.

TriggerBindStarts whendo input
HTTPon(http.get(), flow)A requestMerged path / query / body
Signalon(signalHandle, flow)fx.emitPayload (schema)
Clockon(clockDecl, flow)Scheduler ticknone (_)
CDCon(db.table(t).changed(), flow)Committed SQL write{ before, after, table, action, id }
Call-onlycall("name", { in, out, do, … })fx.callCallee in
MCPon(mcp.tool("x", { in, out }).gate(…), flow)MCP tools/callTool args

signal.live is an HTTP SSE tape — bind it with http.live, not as a worker.

The Capabilities of Flow

HTTP

REST verbs, RFC 10008 QUERY, CRUD mounts, and live SSE.

Routing

File-tree stamps for HTTP paths, Flow names, and client units.

Consumers

Signal workers, named Clock jobs, and SQL CDC — one species.

Durable Workflows

fx.step replay, LIFO undo, durable sleep, and crash resume.

Options Reference

Second argument to flow(name, options) — or the only argument to a nameless flow({ do }). Invoke contracts (in / out / errors / breaking) belong on the exposure — see Contracts below.

OptionTypeDefaultMeaning
do(input, fx) => output | FlowFailure(required)Handler. Missing do throws flow() expected an options bag with a do handler
durablebooleanfalseJournal fx.step / sleep / gated fx calls
retryFxRetryOptionsomittedWhole-do retry on throw (same journal when durable)
cacheboolean | stringomitted (auto)Read-only Flows cache automatically; false opts out; "30s" adds TTL
compensate(ctx, fx) => unknownomittedAfter LIFO { undo }, before the run commits failed
plane"user" | "operator""user"Operator bypasses RLS; user must not fx.call operator
effectsEffectsinferredCapability token — write this only when inference cannot see the body
slo{ availability?, latency? }omittedManifest metadata (Console / docs)
tenantScopedbooleantrue when tenancy is onfalse skips tenant-role scope union

Consequence: durable: true disables automatic read-cache for that Flow.

Contracts

Detailed section

Invoke contracts live on the exposure — http.post({ in, out, errors }), call("name", { in, out, do }), or mcp.tool("x", { in, out }). The Manifest still shows flat flows.*.{in,out,errors,breaking} as a projection from that exposure. in runs before do; out runs after a successful return. fx.fail skips out. Signal / Channel schema is a separate emit contract (validated at fx.emit / fx.send).

Any library with ~standard (Standard Schema V1) works. Zod is the usual choice:

import { on, flow, http } from "okengine";
import { z } from "zod";

on(
  http.post({
    in: z.object({ sku: z.string(), qty: z.number().int().min(1) }),
    out: z.object({ id: z.string() }),
  }),
  flow({
    do: async (input, fx) => ({ id: fx.id() }),
  }),
);

Valibot (v.object) and ArkType (type({…})) bind the same way. Shared DTOs belong in shapes.ts next to the unit — that filename is never a route.

The fx door

Detailed section

If you only need store / emit, jump to the table. fx is the only I/O surface inside do. The compiler records what you touch as effects on the Manifest.

CallRecordsUse
fx.store(db)reads / writes sql:…SQL (and other Store facets)
fx.emit(signal, payload)emitsSignal outbox
fx.send(template, opts)sendsChannel template
fx.ask(prompt, opts)asksAI prompt
fx.vault.get(secret)secretsDeclared secret (never a raw value in source)
fx.call(flow, input?)callsAnother Flow — waits for return
fx.id()—OKID — 21-char native id from okengine/okid
fx.clock.now()—Deterministic time
fx.fail(code, data)—Typed failure value
fx.step(name, fn)journalDurable checkpoint

Call-only

Detailed section

Prefer call("name", { in, out, do, … }). internal exists so call-only is a trigger value — on(internal, flow) — when you need all kinds addressable the same way.

src/flows/orders/checkout.ts
import { on, flow, http } from "okengine";
import { z } from "zod";
import { chargeCard } from "@/flows/payments/charge";

export const checkout = on(
  http.post({ in: z.object({ sku: z.string() }) }),
  flow({
    do: async ({ sku }, fx) => {
      const { chargeId } = await fx.call(chargeCard, { amount: 50 });
      return { sku, chargeId };
    },
  }),
);

Do not fx.call a Flow that parks on fx.clock.sleep — the caller receives undefined and continues; the child wakes later as its own run. Sleep on the root durable Flow, or split with fx.emit. See Workflows.

Cache, retry, and plane

Read-only Flows (Store reads, no writes, no asks, not durable) cache automatically. No cache: option required. Mutations and durable: true stay uncached:

flow("catalog.get", {
  cache: "30s",
  do: async ({ id }, fx) => {
    const [row] = await fx.store(db).select().from(products).where(eq(products.id, id));
    return row;
  },
});

cache: false opts out. A duration string adds TTL on top of write invalidation.

Troubleshooting

Learn more

  • The Architecture — on, trigger, flow, do, fx
  • HTTP — verbs, envelopes, resources, live SSE
  • Routing — file-tree stamps, barrels, OKE1030 · OKE1040–1045
  • Consumers — Signal / Clock / CDC
  • Workflows — durable: true + fx.step
  • fx — every method inside do
  • Gate — .gate(...) / .public() on the trigger
  • Errors — OKE1001–1009 · OKE1020 · ValidationError · denials
  • MCP — mcp.tool exposure

Next

HTTP

Synchronous REST, QUERY, resources, and live SSE.

Consumers

Signal workers, named Clock jobs, and SQL CDC — one Flow species.

Durable Workflows

Step journaling and multi-step distributed execution.

The Architecture

Five pieces behind on(trigger, flow).

Elements

The closed set of eight semantic primitives forming the entire OKEngine backend model.

HTTP

Synchronous REST endpoints, RFC 10008 QUERY, CRUD mounts, live SSE streams, and gate chains on Flow.

On this page

Smallest ExampleDefine a FlowCall itProgressive PatternsTrigger ReferenceThe Capabilities of FlowOptions ReferenceContractsThe fx doorCall-onlyCache, retry, and planeTroubleshootingLearn moreNext