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

<Callout title="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.
</Callout>

<FlowShape />

## Smallest Example

<Steps>

<Step>
### Define a Flow

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

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

</Step>

<Step>
### Call it

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

Response:

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

</Step>

</Steps>

<Callout title="Omit path and name">
  Tree default: `http.get()` + `flow({ do })` — no path or name strings. Pass
  either only for
  [control](/docs/elements/flow/routing#when-to-omit--when-to-pass).
</Callout>

<Callout title="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)`.
</Callout>
## Progressive Patterns

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

<Tabs items={["Minimal", "Validated", "Failures", "Call-only"]}>

<Tab value="Minimal">

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

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

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

</Tab>

<Tab value="Validated">

`in` / `out` live on the HTTP bag. Invalid input never enters `do`:

```typescript title="src/flows/notes/create.ts"
import { on, flow, http } from "okengine";
import { z } from "zod";

export const create = on(
  http.post({
    in: z.object({ title: z.string().min(1) }),
    out: z.object({ id: z.string(), title: z.string() }),
  }),
  flow({
    do: async ({ title }, fx) => {
      const id = fx.id();
      return { id, title };
    },
  }),
);
```

</Tab>

<Tab value="Failures">

Declare domain errors on the exposure bag and return `fx.fail` — do not throw for expected failures:

```typescript title="src/flows/orders/[id]/get.ts"
import { on, flow, http } from "okengine";
import { z } from "zod";
import { eq } from "drizzle-orm";
import { db, orders } from "@/schema";

export const get = on(
  http.get({
    in: z.object({ id: z.string() }),
    out: z.object({ id: z.string(), sku: z.string(), qty: z.number() }),
    errors: { NotFound: z.object({ id: z.string() }) },
  }),
  flow({
    do: async ({ id }, fx) => {
      const [order] = await fx.store(db).select().from(orders).where(eq(orders.id, id));
      if (!order) return fx.fail("NotFound", { id });
      return order;
    },
  }),
);
```

</Tab>

<Tab value="Call-only">

Use `call(name, { … })` — contract and `do` on one bag. The parent records `calls: ["payments.charge"]`.
`fx.call` waits for the callee:

```typescript title="src/flows/payments/charge.ts"
import { call } from "okengine";
import { z } from "zod";
import { db, charges } from "@/schema";

export const chargeCard = call("payments.charge", {
  in: z.object({ amount: z.number() }),
  out: z.object({ chargeId: z.string() }),
  do: async ({ amount }, fx) => {
    const chargeId = fx.id();
    await fx.store(db).insert(charges).values({ id: chargeId, amount });
    return { chargeId };
  },
});
```

```typescript
const { chargeId } = await fx.call(chargeCard, { amount: 50 });
```

</Tab>

</Tabs>

## Trigger Reference

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

<FlowTriggers />

| Trigger   | Bind                                           | Starts when         | `do` input                             |
| --------- | ---------------------------------------------- | ------------------- | -------------------------------------- |
| HTTP      | `on(http.get(), flow)`                         | A request           | Merged path / query / body             |
| Signal    | `on(signalHandle, flow)`                       | `fx.emit`           | Payload (`schema`)                     |
| Clock     | `on(clockDecl, flow)`                          | Scheduler tick      | none (`_`)                             |
| CDC       | `on(db.table(t).changed(), flow)`              | Committed SQL write | `{ before, after, table, action, id }` |
| Call-only | `call("name", { in, out, do, … })`             | `fx.call`           | Callee `in`                            |
| MCP       | `on(mcp.tool("x", { in, out }).gate(…), flow)` | MCP `tools/call`    | Tool args                              |

`signal.live` is an HTTP SSE tape — bind it with [`http.live`](/docs/elements/flow/http#live-streams), not as a worker.

## The Capabilities of Flow

<Cards>
  <Card
    title="HTTP"
    description="REST verbs, RFC 10008 QUERY, CRUD mounts, and live SSE."
    href="/docs/elements/flow/http"
  />
  <Card
    title="Routing"
    description="File-tree stamps for HTTP paths, Flow names, and client units."
    href="/docs/elements/flow/routing"
  />
  <Card
    title="Consumers"
    description="Signal workers, named Clock jobs, and SQL CDC — one species."
    href="/docs/elements/flow/consumers"
  />
  <Card
    title="Durable Workflows"
    description="fx.step replay, LIFO undo, durable sleep, and crash resume."
    href="/docs/elements/flow/workflows"
  />
</Cards>

## 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](#contracts) below.

| Option         | Type                                   | Default                   | Meaning                                                                         |
| -------------- | -------------------------------------- | ------------------------- | ------------------------------------------------------------------------------- |
| `do`           | `(input, fx) => output \| FlowFailure` | _(required)_              | Handler. Missing `do` throws `flow() expected an options bag with a do handler` |
| `durable`      | `boolean`                              | `false`                   | Journal `fx.step` / sleep / gated `fx` calls                                    |
| `retry`        | `FxRetryOptions`                       | omitted                   | Whole-`do` retry on throw (same journal when durable)                           |
| `cache`        | `boolean \| string`                    | omitted (auto)            | Read-only Flows cache automatically; `false` opts out; `"30s"` adds TTL         |
| `compensate`   | `(ctx, fx) => unknown`                 | omitted                   | After LIFO `{ undo }`, before the run commits `failed`                          |
| `plane`        | `"user" \| "operator"`                 | `"user"`                  | Operator bypasses RLS; user must not `fx.call` operator                         |
| `effects`      | `Effects`                              | inferred                  | Capability token — write this only when inference cannot see the body           |
| `slo`          | `{ availability?, latency? }`          | omitted                   | Manifest metadata (Console / docs)                                              |
| `tenantScoped` | `boolean`                              | `true` when tenancy is on | `false` skips tenant-role scope union                                           |

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

## Contracts

<Callout title="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`).
</Callout>

<Tabs items={["Standard Schema", "Failures", "Envelope"]}>

<Tab value="Standard Schema">

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

```typescript
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.

</Tab>

<Tab value="Failures">

Errors at the Flow boundary are **values**. Throw only for bugs. Declare `errors` on the
exposure and return from `do`:

```typescript
call("orders.create", {
  in: z.object({ sku: z.string(), qty: z.number().int().min(1) }),
  out: z.object({ id: z.string() }),
  errors: {
    OutOfStock: z.object({ available: z.number() }),
  },
  do: async (input, fx) => {
    const [row] = await fx.store(db).select().from(stock).where(eq(stock.sku, input.sku));
    if (!row || row.available < input.qty) {
      return fx.fail("OutOfStock", { available: row?.available ?? 0 });
    }
    return { id: fx.id() };
  },
});
```

`fx.fail(code, data, { message? })` builds `{ data: null, error: { code, data, message? } }`.
The typed client narrows on `res.error.code`.

</Tab>

<Tab value="Envelope">

HTTP success from a returned value is `200` + `{ data, error: null }`. `undefined` is
`204` with an empty body. Typed failures use `{ data: null, error }`:

```json
{
  "data": null,
  "error": {
    "code": "OutOfStock",
    "data": { "available": 0 }
  }
}
```

Status for `error.code`:

| Code                                                  | Status |
| ----------------------------------------------------- | ------ |
| `ValidationError`                                     | `422`  |
| `Unauthorized`                                        | `401`  |
| `Forbidden`                                           | `403`  |
| `RateLimited`                                         | `429`  |
| Any other declared code (`NotFound`, `OutOfStock`, …) | `400`  |

A bare `404` with body `Not Found` means **no route matched** — not `fx.fail("NotFound")`.

</Tab>

</Tabs>

<Accordions>

<Accordion title="store.resource schemas">
  `store.resource(db, table, { in, out })` requires `in` (create body) and `out` (item
  shape). List / get / update / remove Flows are built for you — contracts are stamped
  from the resource factory. Handwritten invoke contracts go on `http.*` / `call` /
  `mcp.tool` — see [HTTP · Resources](/docs/elements/flow/http#resources).
</Accordion>

<Accordion title="ValidationError payload">
  Failed `in` (or `out`) is `ValidationError` with `error.data.issues` — each issue has `message`
  and `path`. HTTP status is **422**. The handler never ran.
</Accordion>

<Accordion title="Name stamping">
  Prefer nameless `flow({ do })` on HTTP tree files — `src/flows/notes/[id]/get.ts` +
  `export const get` stamps `notes.get`. Signal / Clock workers pass `flow("name", { do })`
  (**OKE1072**; Clock inline is [Clock · Inline or named export](/docs/elements/clock#inline-or-named-export)).
</Accordion>

</Accordions>

## The fx door

<Callout title="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.
</Callout>

| Call                       | Records                    | Use                                           |
| -------------------------- | -------------------------- | --------------------------------------------- |
| `fx.store(db)`             | `reads` / `writes` `sql:…` | SQL (and other Store facets)                  |
| `fx.emit(signal, payload)` | `emits`                    | Signal outbox                                 |
| `fx.send(template, opts)`  | `sends`                    | Channel template                              |
| `fx.ask(prompt, opts)`     | `asks`                     | AI prompt                                     |
| `fx.vault.get(secret)`     | `secrets`                  | Declared secret (never a raw value in source) |
| `fx.call(flow, input?)`    | `calls`                    | Another 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)`        | journal                    | Durable checkpoint                            |

<Accordions>

<Accordion title="Side channels">
  `Date.now()`, `new Date()`, `setTimeout`, global `fetch`, and `node:fs` skip the ledger.
  Tests cannot time-travel; durable replay cannot skip the work; cache cannot see the read.

**Fix:** `fx.clock.now()`, `fx.store`, `fx.send`, or wrap a provider in `fx.step`.

</Accordion>

<Accordion title="fx.call identity">
  `fx.call` starts the callee with an **empty** `fx.auth` (fail-closed). `fx.tenant.id` propagates.
  For audit only, read `fx.principal` — gates never consult it. See [fx](/docs/reference/fx).
</Accordion>

<Accordion title="Undeclared effects (OKE1001–1007)">
  Explicit `effects` that drift from the body throw at runtime (`Flow "{flow}" writes "{resource}"
  without declaring it.`). Most apps never write `effects` — inference covers them. **OKE1020** is
  deploy-shaped boot with neither inference nor a block.
</Accordion>

</Accordions>

## Call-only

<Callout title="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.
</Callout>

```typescript title="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](/docs/elements/flow/workflows).

## Cache, retry, and plane

<Tabs items={["Cache", "Retry", "Plane"]}>

<Tab value="Cache">

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

```typescript
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.

</Tab>

<Tab value="Retry">

`flow({ retry })` re-enters the whole `do`. `retries` is extra attempts after the first.
Prefer `fx.retry` **inside** `fx.step` so a completed charge never re-runs:

```typescript
flow({
  retry: { retries: 3, delay: "200ms", backoff: 2 },
  do: async (input, fx) => {
    return await fx.ask(flakyModel, { prompt: input.text });
  },
});
```

| `retry` option | Default       | Meaning                            |
| -------------- | ------------- | ---------------------------------- |
| `retries`      | `0`           | Extra attempts after the first     |
| `delay`        | `50` (ms)     | Initial wait (`"200ms"` allowed)   |
| `backoff`      | `2`           | Multiplier after each retry        |
| `jitter`       | `true`        | Full jitter                        |
| `when`         | thrown errors | Skips abort and durable-sleep park |

</Tab>

<Tab value="Plane">

`"user"` is the application default. `"operator"` is Console — RLS is bypassed,
`fx.operator` is the principal, `fx.auth` must not appear in that body:

```typescript
flow("ops.allOrders", {
  plane: "operator",
  do: async (_, fx) => {
    return await fx.store(db).select().from(orders);
  },
});
```

A user-plane Flow that `fx.call`s an operator Flow fails compile:
`cross-plane call: user flow "…" calls operator flow "…"`.

</Tab>

</Tabs>

<Accordions>

<Accordion title="compensate context">
  Durable-only. Runs after reverse `{ undo }`, never on success, retry attempts, or sleep
  park. Context: `{ input, error, completedSteps }` — forward step names only. Full
  physics: [Workflows · Compensation](/docs/elements/flow/workflows#compensation).
</Accordion>

<Accordion title="breaking and tenantScoped">
  `breaking: true` on the exposure bag lets `oke doctor --diff` accept that Flow's contract
  break. It does not cover a different Flow.

`tenantScoped: false` skips tenant-role scope union even when `fx.tenant.id` is set.
Default is `true` once `gate.auth.tenant` is on.

</Accordion>

<Accordion title="Hooks and plugins">
  `flowDef.hook(stage, fn)` registers a per-Flow hook (`onRequest` · `onParse` · `onAuth` ·
  `beforeHandle` · `afterHandle` · `onError` · `onResponse`). `flowDef.plug(plugin)` scopes a plugin
  to that Flow — see [Plugins](/docs/reference/plugins).
</Accordion>

</Accordions>

## Troubleshooting

<Accordions>

<Accordion title='TypeError: flow() expected an options bag with a do handler'>
  `flow()` requires `{ do }`. `flow("name")` with no options, or a bag without `do`,
  throws at declaration — before `on()`.
</Accordion>

<Accordion title="TypeError: on() expected a trigger or signal handle">
  First argument must be an HTTP trigger, Signal handle, Clock handle, `db.table(…).changed()`,
  `internal`, or `mcp.tool(…)`. A bare interval string is not a trigger — wrap it in
  `clock.every("name", "1h")`.
</Accordion>

<Accordion title="TypeError: on() expected a flow() definition as the second argument">
  Second argument must be the object `flow()` returned. Passing a plain function or
  forgetting `flow({ do })` fails here. Resource mounts take **no** second argument —
  see [HTTP · Resources](/docs/elements/flow/http#resources).
</Accordion>

<Accordion title="Direct Date.now, fetch, or node: import inside do">
  Side channels skip effect tracking and durable replay.

**Fix:** `fx.clock.now()`, `fx.store`, `fx.send`, or `fx.step` around the provider.

</Accordion>

<Accordion title="Thrown Error becomes a mystery 500 instead of a typed envelope">
  Uncaught exceptions are defects. Declare the code in `errors` on the exposure and `return
  fx.fail("OutOfStock", payload)` from `do`.
</Accordion>

<Accordion title="422 ValidationError — path param missing from in">
  `http.get("/users/:id")` merges `{id}`. A schema that expects `userId` fails before `do`. Align
  path keys with `in` object keys. See [HTTP · Request
  Parsing](/docs/elements/flow/http#request-parsing).
</Accordion>

<Accordion title="fx.fail('NotFound') is 400, not 404">
  Custom domain codes map to **400**. A bare `404` `Not Found` means the router found no method +
  path. Use `fx.fail` for domain misses; fix the route for missing bindings.
</Accordion>

<Accordion title="Read-only Flow never cache-hits">
  Auto-cache needs Store `reads`, no `writes`, no `asks`, and `durable` off. Empty effect sets stay
  uncached. Opt in with a duration (`cache: "30s"`) only after a real read is inferred — or pass
  `cache: false` to disable.
</Accordion>

<Accordion title='cross-plane call: user flow "…" calls operator flow "…"'>
  User-plane Flows cannot `fx.call` operator Flows. Keep operator work on `plane: "operator"` and
  invoke it from Console, or split a user-safe callee.
</Accordion>

<Accordion title="OKE1020 — no declared effects">
  Cause: `Flow "{flow}" has no declared effects and no Manifest to derive them from.` Extract
  failures append `Manifest extract failed — …`. Boot with `oke dev` / `oke build`; install
  `oxc-parser` if extract cannot load. On Windows use `bun run dev` / `bunx oke dev`.
</Accordion>

</Accordions>

## Learn more

- [The Architecture](/docs/understand/the-architecture) — `on`, trigger, `flow`, `do`, `fx`
- [HTTP](/docs/elements/flow/http) — verbs, envelopes, resources, live SSE
- [Routing](/docs/elements/flow/routing) — file-tree stamps, barrels, OKE1030 · OKE1040–1045
- [Consumers](/docs/elements/flow/consumers) — Signal / Clock / CDC
- [Workflows](/docs/elements/flow/workflows) — `durable: true` + `fx.step`
- [fx](/docs/reference/fx) — every method inside `do`
- [Gate](/docs/elements/gate) — `.gate(...)` / `.public()` on the trigger
- [Errors](/docs/reference/errors) — OKE1001–1009 · OKE1020 · ValidationError · denials
- [MCP](/docs/elements/ai/mcp) — `mcp.tool` exposure

## Next

<Cards>
  <Card
    title="HTTP"
    description="Synchronous REST, QUERY, resources, and live SSE."
    href="/docs/elements/flow/http"
  />
  <Card
    title="Consumers"
    description="Signal workers, named Clock jobs, and SQL CDC — one Flow species."
    href="/docs/elements/flow/consumers"
  />
  <Card
    title="Durable Workflows"
    description="Step journaling and multi-step distributed execution."
    href="/docs/elements/flow/workflows"
  />
  <Card
    title="The Architecture"
    description="Five pieces behind on(trigger, flow)."
    href="/docs/understand/the-architecture"
  />
</Cards>