`fx` is the second argument of every `do` — the single door to the world. This page is the whole surface; each entry notes the **effect it records**, which feeds the Manifest, caching, and capability checks. All world access goes through `fx`. No `node:` I/O, no raw `fetch` for side effects (use `fx.fetch` instead), no `Date.now()` — clocks, stores, channels, vault, AI, and outbound HTTP only through this object. ## Smallest Example ### Use `fx` inside `do` ```typescript title="src/flows/notes/create.ts" import { on, flow, http } from "okengine"; import { z } from "zod"; import { db, notes } from "@/schema"; export const create = on( http.post({ in: z.object({ title: z.string().min(1) }), }), flow({ do: async ({ title }, fx) => { const id = fx.id(); await fx.store(db).insert(notes).values({ id, title }); return fx.json.create({ id, title }); }, }), ); ``` ### Type extracted helpers as `Fx` ```typescript import type { Fx } from "okengine"; async function loadNote(id: string, fx: Fx) { return fx.store(db).findById(notes, id); } ``` A narrower structural type will not match `store()` overloads. See [Flow](/docs/elements/flow). ## What the effects ledger powers | Derived behavior | From | | ---------------------- | -------------------------------------- | | Cache invalidation | Inferred / ledgered `reads` · `writes` | | Live queries / CDC | Store `writes` observed as row events | | Least-privilege tokens | Effect matrices on Flows | | Deterministic tests | Injectable clock · store · channel | | Runs observability | Wide events per invocation (`fx.runs`) | ## Stores | Signature | Records | Returns | | ------------------------------------------------------------------- | -------------------------- | ------------------------------------------------------------------------------------------- | | `fx.store(sqlDecl).select().from(t)…` | `read` | inferred rows (`where` · `orderBy` · `limit` · `offset` chainable) | | `fx.store(sqlDecl).insert(t).values(v)` | `write` | `Promise` | | `fx.store(sqlDecl).update(t).set(v).where(…)` | `write` | `Promise` | | `fx.store(sqlDecl).delete(t).where(…)` | `write` | `Promise` | | `fx.store(sqlDecl).findById(t, id)` | `read` | row \| undefined | | `fx.store(sqlDecl).search(table, { query, fuse?, rerank?, … })` | `read` (+ `ask` if rerank) | `{ data, meta }` hybrid BM25 ± LSH — see [Search](/docs/elements/store/search) | | `fx.store(kv).get / set(key, value, ttl?) / delete / list(prefix?)` | read / write | per op | | `fx.store(files).put / get / delete / list(prefix?)` | read / write | per op | | `fx.store(files).image(key\|bytes).…` | read / write | Bun.Image chain; terminals gate (see [Store](/docs/elements/store#images--image--putimage)) | | `fx.store(files).putImage(key, data, opts?)` | `write` | original + variants (+ optional LQIP) | | `fx.store(index).upsert / search(vector, topK?) / delete` | read / write | per op | See [Store](/docs/elements/store) for the query-builder surface. ## Signals | Signature | Records | Notes | | ------------------------------------- | ---------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `fx.emit(signal, payload?, { key? })` | `emit` | Pass a `SignalDecl` handle to type-check `payload`; a string name stays `unknown` (runtime `schema` still applies). Commits the signal outbox when the call resolves; optional `key` serializes `once` per key; stamps producer run id as `parentRunId` for trace chains; throws **OKE1240** (orphan) or **OKE1250** (schema) | | `fx.deadLetters(signal)` | `read` `signal:` | Dead-lettered messages for that signal. Payload typed from `SignalDecl`. Page with `fx.json.withQuery`. Cross-signal throws **OKE1001**. | | `fx.live(signal, { match? })` | `read` `signal:` | Live tape as SSE. Returns `JsonStreamResult` (object chunks, `id:` on the wire). Cross-signal throws **OKE1001**. Do not wrap with `fx.json.stream`. | ## Runs (observability read) Declare `effects: { reads: ["runs"] }`. Powers native SLO checkers (Clock + Channel) without a parallel metrics API. | Signature | Records | Returns | | ---------------------------------------- | ------------- | ---------------------------------------------------------------------------------------- | | `fx.runs.query(sql)` | `read` `runs` | SQL rows (`FROM runs` on files/memory). Unrestricted Flow SQL — not the Console sandbox. | | `fx.runs.all()` | `read` `runs` | All wide events | | `fx.runs.window(flow, windowMs?)` | `read` `runs` | Rolling P50/P95/P99 + error rate (default 5m) | | `fx.runs.checkSlo(flow, slo, windowMs?)` | `read` `runs` | Availability / latency breaches | ```typescript const sloCheckClock = clock.every("ops.slo-check", "5m"); on( sloCheckClock, flow("ops.slo-check", { effects: { reads: ["runs"], sends: ["slo-alert"] }, do: async (_, fx) => { const breaches = await fx.runs.checkSlo( "checkout.create", { availability: "99.9%", latency: { p95: "200ms" } }, 5 * 60_000, ); if (breaches.length === 0) return; await fx.send(sloAlert, { to: "oncall@example.com", data: { flow: "checkout.create", count: breaches.length }, }); }, }), ); ``` ## Flows | Signature | Records | Returns / notes | | ------------------------------------------------------------------------------ | ----------------------- | -------------------------------------------------------------------------------------------------- | | `fx.call(flow, input?)` | `call` | The callee's `out` — runs through the same pipeline | | `fx.step(name, fn)` | — | Durable step: replays from the journal, never re-runs | | `fx.all([...thunks])` | — | Parallel; first rejection aborts siblings | | `fx.race([...thunks])` | — | First settle wins; losers aborted | | `fx.retry(fn, opts?)` | — | Exponential backoff + jitter (plain Promise) | | `fx.using(acq, rel, use)` | — | `release` runs once on settle or ambient abort | | `fx.signal` | — | Ambient `AbortSignal` for the current branch | | `fx.fail(code, data, opts?)` | — | Typed failure value (`opts.message` overrides) | | `fx.auth.createApiKey({ name, scopes, expiresIn?, ipAllowlist?, rateLimit? })` | `write` `auth:api-keys` | Secret once. Creator is live `userId` / `scopes`. Session only. `ipAllowlist` is IPs or hostnames. | | `fx.auth.listApiKeys()` | `read` `auth:api-keys` | Keys this session minted | | `fx.auth.revokeApiKey(id)` | `write` `auth:api-keys` | Owner only | | `fx.auth.rotateApiKey(id)` | `write` `auth:api-keys` | New secret once. Owner only | | `fx.auth.updateApiKey(id, …)` | `write` `auth:api-keys` | Name / scopes / expiry / allowlist / rate. Re-attenuates | | `fx.auth.listTenants()` | `read` `auth:tenants` | Memberships for the live session. Session only | | `fx.auth.switchTenant(id)` | `write` `auth:tenants` | New access+refresh, new family, `tid` on both. Never Set-Cookie. Session only | | `fx.auth.createTenant({ name, slug?, id? })` | `write` `auth:tenants` | Creator becomes a member. Session only | | `fx.auth.upsertTenantRole({ tenantId, roleName, scopes })` | `write` `auth:tenants` | Application scopes only — `console:*` is unknown_scope | `fx.call` starts the callee with an **empty** `fx.auth` (fail-closed for authorization) and propagates `fx.tenant.id`. For audit/attribution only, read `fx.principal` — it propagates the originating identity without copying into `fx.auth`. Gates never consult `fx.principal`. ## Concurrency and retry Pass **thunks** to `all` / `race` — not already-started Promises — so each branch gets an abort scope before work begins. ```typescript const [user, stock] = await fx.all([ () => fx.store(db).findById(users, input.userId), () => fx.store(db).findById(inventory, input.sku), ]); const charge = await fx.step("charge", () => fx.retry(() => fx.call(stripeCharge, { amount: input.total }), { retries: 3, delay: "100ms", backoff: 2, jitter: true, }), ); ``` | `fx.retry` option | Default | Meaning | | ----------------- | ------- | -------------------------------------------- | | `retries` | `0` | Extra attempts after the first | | `delay` | `50` | Initial backoff — ms number or `"100ms"` | | `backoff` | `2` | Multiplier after each retry | | `jitter` | `true` | Full jitter on the delay (thundering-herd) | | `when` | thrown | Predicate; skips `AbortError` and sleep park | Losing branches see `fx.signal` abort. Drivers that do not yet honor the signal may still finish in the background — check `fx.signal.aborted` in long user work, and prefer `fx.all` over bare `Promise.all`. `fx.using(acquire, release, use)` scopes a process-local resource to one attempt: `release` runs exactly once when `use` settles **or** when the ambient signal aborts (a sibling `fx.race` winner, a failing `fx.all` sibling). It is not journaled — do not hold handles across durable park/resume. ```typescript const rows = await fx.using( () => pool.acquire(), (conn) => conn.release(), (conn) => conn.query("select …"), ); ``` **Consequence:** put `fx.retry` inside `fx.step` on durable flows so a completed charge never re-runs on resume. Coarse whole-body retry is `flow(name, { retry: { … } })` on the same journal session. ## Channel | Signature | Records | Notes | | -------------------------------------------- | ------- | ------------------------------------------------------------------------------------------ | | `fx.send(template, { to?, data?, via?, … })` | `send` | `via` orders fallback; `locale` / `profileLocale` / `acceptLanguage` feed the locale chain | Omit locale opts and the send uses `fx.locale`. Dry runs record _would have fired_ and never contact a provider. Channel bodies use `{{field}}` catalogs — not ICU (see [Channel](/docs/elements/channel)). ## Outbound HTTP | Signature | Records | Notes | | ---------------------- | --------------------------- | ------------------------------------------------------------------------- | | `fx.fetch(url, init?)` | `fetch` on the URL hostname | Always stamps `EffectEntry.external` with `{ host, kind: "third-party" }` | Declare hosts in `effects.fetches` (e.g. `["api.stripe.com"]`). Prefer `fx.step`; use `fx.retry` only when the remote API is safe to repeat. Dry runs never hit the network. Use this for third-party REST outside Channel / AI / Store — not as a substitute for those elements. ## AI | Signature | Records | Returns | | ----------------------------------------------------- | ------------------------- | -------------------------------------------------------------------------------------------------------- | | `fx.ask(prompt, input?, { via?, tools?, maxSteps? })` | `ask` (+ `call` per tool) | Object validated against the prompt's `out` | | `fx.run(agent, input?)` | `ask` | Agent result | | `fx.stream(model, { prompt?, data?, via? })` | `ask` | `AsyncIterable` — real driver stream; cancels via ambient `fx.signal` (HTTP disconnect included) | | `fx.search(embed, query, { topK? })` | `read` | Matches from the index/embed | AI calls are nondeterministic: journaling is forced on and auto-cache disabled around them. `tools` are Flow refs — each model tool call goes through `fx.call` (same capability and Runs path). Driver-reported `EffectEntry.external` marks cloud providers as `third-party` and self-hosted / `provider: "local"` as `infrastructure` (Console waterfall dashed egress + host tooltip). ## Vault | Signature | Records | Returns / notes | | ---------------------------------- | -------- | -------------------------------------------------------------------------------------------------- | | `fx.vault.get(contract)` | `secret` | `Promise>` — prints/logs as a placeholder; `.reveal()` at the credential boundary | | `fx.vault.set(path, value, opts?)` | `secret` | `{ path, version }` — needs a bound Vault backend | | `fx.vault.rotate(path, value)` | `secret` | `{ path, version }` — new version with a fresh data key | | `fx.vault.delete(path)` | `secret` | `boolean` — crypto-shreds the path | | `fx.vault.list(prefix?)` | — | Secret paths, never values | | `fx.vault.status()` | — | `{ sealed, initialized, backend }` | `get` reads through the boot resolution chain. Everything else needs the encrypted-at-rest backend (`drivers.vault = "vault"`) and throws without it. ## Clock | Signature | Notes | | --------------------------------- | ------------------------------------------------------------------- | | `fx.clock.now()` | Epoch-ms, injectable — the only legal "now" | | `fx.clock.ago(duration)` | Instant before now (`"30d"` → now − 30 days) | | `fx.clock.fromNow(duration)` | Instant after now (`"14d"` → now + 14 days) | | `fx.clock.duration(duration)` | Span in ms — offset a stored instant (`createdAt + duration("7d")`) | | `fx.clock.sleep(label, duration)` | Durable sleep in `durable` flows; immediate otherwise | Durations: `"200ms"` · `"30s"` · `"2m"` · `"1h"` · `"7d"`. A `"d"` is 86_400_000 ms, not a calendar day. Unknown strings parse as `0`. ## Cache Read-only flows cache automatically from inferred or ledgered `reads` — no `fx.cache` call and no `cache:` default on the flow. Writes invalidate those keys. Use `cache: false` to opt out, or `cache: "30s"` for a TTL. `fx.cache` is the manual (tier-3) surface: | Signature | Notes | | -------------------------------------- | ----------------------------------------- | | `fx.cache.get(key)` | Value or `undefined` | | `fx.cache.set(key, value, ttl?)` | Optional TTL string | | `fx.cache.getOrSet(key, ttl, produce)` | Read-through; writes invalidate by effect | ## Responses | Helper | Status | Body | | ----------------------------------------- | ------ | -------------------------------------------------------------- | | `fx.json.ok(value, { meta? })` | 200 | `{ data, meta?, error: null }` | | `fx.json.create(value)` | 201 | `{ data, error: null }` | | `fx.json.empty()` | 204 | no body | | `fx.json.with(page)` / `with(data, meta)` | 200 | `{ data, meta, error: null }` — already-built pager | | `fx.json.withQuery(rows, input, spec?)` | 200 | In-memory list page — zero-config `q` / auto-eq / PostgREST | | `fx.json.stream(chunks)` | 200 | `text/event-stream` — JSON `data:` frames, then `data: [DONE]` | | `fx.live(signal)` | 200 | Same SSE carrier for a live signal (payload frames + `id:`) | Returning a plain value instead answers 200 with `{ data: value, error: null }` — the helpers exist for status and `meta` control. Pass `fx.stream(...)` into `fx.json.stream` to reach the HTTP client token-by-token. ## Logging, i18n, ids | Signature | Notes | | ------------------------------------------ | ----------------------------------------------------------- | | `fx.log.debug/info/warn/error(msg, data?)` | Redacting — secrets print as `***` | | `fx.t(key, values?)` | ICU MessageFormat — active locale → `i18n.default` → key | | `fx.locale` | Active locale (`Accept-Language` matched to `i18n.locales`) | | `fx.id()` | OKID — 21-char native id from `okengine/okid` | Catalogs, ICU syntax, and `Register` augmentation live on [i18n](/docs/reference/i18n). Id options: [OKID](/docs/reference/okid). Localized `fx.fail` / `OkeError` catalogs: [Errors](/docs/reference/errors). ## Principals | Property | Shape | | -------------- | ------------------------------------------------------------------------------------- | | `fx.auth` | `{ userId, scopes, verified?, apiKeyId? }` plus key and tenant methods (session only) | | `fx.operator` | `{ id: string \| null }` — Console plane | | `fx.principal` | Read-only origin: `userId`, `operatorId`, `scopes`, `verified?`, `plane?` | | `fx.tenant` | `{ id: string \| null }` — active tenant (propagates on `fx.call`) | **Consequence:** use `fx.auth` / gates for authorization; use `fx.principal` only when a callee must log who started the call chain. A key Bearer sets `userId` to the issuer and `apiKeyId` to the key — see [Gate](/docs/elements/gate#api-keys). ## Not on `fx` Investigated and declined. `fx.runs` already provides per-invocation observability as wide events. Native alerting is `fx.runs` + Clock + Channel — not a second counter/gauge API. Optional OTLP export for existing Grafana/Datadog stacks is additive and never required. ## Troubleshooting An explicit `effects` block drifted from what `do` touches. Add the missing ledger entry (`reads` · `writes` · `emits` · `sends` · `asks` · `embeds` · `secrets` · `calls` · `fetches`), or drop the block so inference covers the Flow ([Errors](/docs/reference/errors)). `fx.emit` with zero subscribers and `optional: false`. Add `on(signal, …)` or declare the signal `optional: true`. Emit payload failed the signal's `schema`. Pass a matching payload or remove the schema. Reading another signal's tape throws **OKE1001**. Declare `effects.reads: ["signal:"]` for that name. Type the parameter as `Fx` from `okengine`, not a hand-rolled structural type. ## Learn more - [Flow](/docs/elements/flow) — why `fx` is the only door - [Channel](/docs/elements/channel) — `fx.send`, consent, locale chain, `{{field}}` catalogs - [i18n](/docs/reference/i18n) — `fx.t`, catalogs, locale matching - [Errors](/docs/reference/errors) — what `fx.fail` produces - [Configuration](/docs/reference/configuration) — drivers and the `i18n` block - [OKID](/docs/reference/okid) — `fx.id()` options ## Next