# AI Flows > Durable multi-step AI pipelines — deterministic or agentic, with human-in-the-loop, chaining, and cadence. Author flows in code (defineFlow) or as data (visual-editor rows); one engine runs both. --- ## AI Flows _Durable multi-step AI pipelines — deterministic or agentic, with human-in-the-loop, chaining, and cadence. Author flows in code (defineFlow) or as data (visual-editor rows); one engine runs both._ `@voltro/plugin-ai-flows` turns a multi-step AI pipeline into a **durable, resumable run**: an orchestrator that chains sub-agents and generation tools (text / image / video / audio) into a bundled result, with **human-in-the-loop** review, **chaining** (flow A feeds flow B), and **cadence** (scheduling). It's the layer above `@voltro/ai` agents — an agent is code; a **flow is data** a non-technical user can author in a visual editor and the framework interprets. Two execution modes: - **Deterministic** (`mode: 'deterministic'`) — run the plan steps in order, verbatim, no orchestrator LLM. Predictable, cheaper, **resumable per step** (a crash/deploy resumes from the first unfinished step via the workflow journal). - **Agentic** (`mode: 'agentic'`) — an orchestrator LLM drives the tools and may reorder/insert/skip the suggested plan, bounded by `maxSteps`. And two authoring **front doors** that lower to one engine: - **Code-first `defineFlow(...)`** — typed, testable, diffable; validates every `{{ref}}` at registration (dangling refs fail at boot, not mid-run). - **Data-driven** — a stored `_voltro_ai_flows` row (from a visual editor); hot-editable, no deploy. ## Install ```ts // app.config.ts import { aiFlowsPlugin } from '@voltro/plugin-ai-flows' export default { plugins: [aiFlowsPlugin()] } // contributes _voltro_ai_flows + _voltro_ai_flow_runs ``` Three tunables, all with defaults — set them when the defaults don't match your process: ```ts // app.config.ts import { aiFlowsPlugin } from '@voltro/plugin-ai-flows' export default { plugins: [ aiFlowsPlugin({ // How long a `human` step parks before the run fails. Default 7 days; // `0` = wait forever. A flow's `humanTimeoutMs` and a step's own // `timeoutMs` both outrank this. humanReviewTimeoutMs: 14 * 24 * 60 * 60 * 1000, // How long a FINISHED run is kept. Default 90 days. runsTtlMs: 180 * 24 * 60 * 60 * 1000, // Ceiling on rows one `/flows` or `/runs` inspect call returns. inspectPageMax: 200, }), ], } ``` Each also has a deploy-time env override, for an operator who can't edit source: `VOLTRO_AI_FLOW_HUMAN_REVIEW_TIMEOUT_HOURS` and `VOLTRO_AI_FLOW_RUNS_TTL_HOURS`. ## Author a flow (code-first) ```ts import { defineFlow, flowStep } from '@voltro/plugin-ai-flows' defineFlow({ name: 'blog-article', brief: [{ key: 'topic', label: 'Topic', type: 'text', required: true }], steps: [ flowStep.agent({ agentRef: 'writer', prompt: 'Draft a post about {{topic}}', outputKey: 'draft' }), flowStep.human({ reviewMode: 'approve', prompt: 'Approve this draft?' }), flowStep.media('image', { prompt: 'Hero image for: {{draft}}', outputKey: 'hero' }), flowStep.note({ prompt: '# {{topic}}\n\n{{draft}}\n\n![hero]({{hero}})', outputKey: 'final' }), ], }) ``` Each step writes `context[outputKey]`; later steps reference it as `{{outputKey}}` (alongside `{{briefKey}}`). The **result bundle** is `outputKey → value`. ## Step kinds - **`agent`** — delegate to a sub-agent (its system prompt + model), text-only. - **`generate`** — `text` / `image` / `video` / `audio` via `@voltro/ai`. - **`structured`** — a JSON object, validated against the step's JSON Schema. - **`note`** — the interpolated prompt itself is the output (no model call). - **`human`** — pause for review (`approve` / `choice` / `text`). ### `structured` — the schema is real A `structured` step's `schema` is a JSON Schema, and it is **both** what the model is told to emit and what the result is validated against: ```ts import { flowStep } from '@voltro/plugin-ai-flows' flowStep.structured({ prompt: 'Extract the invoice fields from: {{document}}', outputKey: 'invoice', schema: { type: 'object', required: ['total', 'currency'], properties: { total: { type: 'number', description: 'Gross total' }, currency: { type: 'string', enum: ['EUR', 'USD'] }, lines: { type: 'array', items: { type: 'string' } }, }, }, }) ``` Objects (with `required`), arrays, the four scalars, `enum`, `const`, both nullability spellings and `anyOf` / `oneOf` unions are modelled; `description` / `title` ride along as annotations. A construct the adapter does not model (`$ref`, `allOf`, …) degrades to "unknown" for **that node** rather than failing the step, and a step with no `schema` behaves as it always did. ## Human-in-the-loop A `human` step sets the run to `waiting` and **parks the workflow without holding a worker** (the durable suspend). The user answers via `respondToFlow`; a reject ends the run cancelled, otherwise the answer lands in `context` and the run resumes. Thousands of runs can wait on review for days at no runtime cost. **Several reviews in one flow are independent.** Each human step parks on its own durable signal, derived from that step's position in the plan the run's journal pinned at step 0 — so the identity is the same on every replay, and answering the first review cannot resolve the second. `respondToFlow` reads the parked step off the run row's live timeline and addresses that step, which is why the UI never has to track a step index; it returns the `stepIndex` it answered. ### The park is bounded, and the bound is yours An approval left over a weekend used to fail the whole flow: the park inherited `@voltro/workflow`'s 24-hour default, which no flow author chose or could change. It resolves most-specific-first now — **step → flow → plugin option → env → 7 days** — and `0` at any level means *wait forever* (the park is slot-free, so an unbounded wait costs no worker): ```ts import { defineFlow, flowStep } from '@voltro/plugin-ai-flows' defineFlow({ name: 'contract-review', // Flow-wide default for every human step that sets none. humanTimeoutMs: 30 * 24 * 60 * 60 * 1000, steps: [ flowStep.human({ reviewMode: 'approve', prompt: 'Legal sign-off?' }), // This one is the CFO's, and it waits as long as it takes. flowStep.human({ reviewMode: 'approve', prompt: 'Budget sign-off?', timeoutMs: 0 }), ], }) ``` A flow authored as DATA carries the same setting in `_voltro_ai_flows.humanTimeoutMs`. When the bound does expire, the run row is written `status: 'failed'` naming the bound that elapsed — it no longer dies with the row still reading `waiting`. ## Conditional steps and fan-out ```ts defineFlow({ name: 'campaign', brief: [{ key: 'mode', label: 'Mode', type: 'text' }], steps: [ flowStep.text({ prompt: 'Draft the copy', outputKey: 'copy' }), // Runs only when the brief asked for the full treatment. flowStep.text({ prompt: 'Write a long-form variant of {{copy}}', outputKey: 'longform', when: { ref: 'mode', op: 'eq', value: 'full' }, }), // These two run CONCURRENTLY — same group, consecutive. flowStep.media('image', { prompt: 'Hero image for {{copy}}', outputKey: 'hero', group: 'assets' }), flowStep.media('image', { prompt: 'Square crop for {{copy}}', outputKey: 'square', group: 'assets' }), // Both group outputs are available again after the group completes. flowStep.note({ prompt: 'Ship {{hero}} and {{square}}', outputKey: 'summary' }), ], }) ``` ### `when:` — run a step only sometimes A step with a `when:` runs only if the condition holds against the run context (brief fields ⊕ earlier `outputKey`s). A step whose condition is false is **skipped**, not failed: it produces no output, so anything referencing it sees an absent value — which is what makes `when` compose with the dependency guard instead of fighting it. The run timeline shows the step as `skipped` with the rendered reason (`{{mode}} equals "full"`), so a step that vanished is never indistinguishable from a step nobody declared. | `op` | True when | |---|---| | `truthy` / `falsy` | the value is present and not `false` / an empty array | | `eq` / `neq` | the value equals the literal (structural for objects and arrays) | | `contains` | the array contains the value, or the string contains the substring | Two deliberate choices, because both look like bugs until you know why: - **A condition is structured data, not an expression string.** A flow can be authored as a stored row a user edits in a browser, and an expression there would be an evaluator running user-authored source on your server. It also lets the boot-time validator refuse a condition on a key nothing produces — the alternative silently skips its step on every run, forever. - **`0` and `''` are TRUTHY here.** A step gated on a generated count or string means "did the producer run", not "is it non-zero". The second is `{ op: 'neq', value: 0 }`, which you can say when you mean it. A compound condition needs two steps, or a `structured` step that computes the boolean. That is a real limit of a parser-free design. ### `group:` — run steps concurrently **Consecutive** steps sharing a `group` name run at the same time. Each keeps its own durable step, so a replay resolves every branch from the journal exactly as it would sequentially — the concurrency is in the execution, not in the durability. Three rules, all enforced at registration rather than at run time: - Steps in one group **cannot read each other's outputs** — they have no order between them. Their outputs become available to everything *after* the group. - A group must be **contiguous**. A group name that stops and resumes would execute as two sequential fan-outs, which is the opposite of what it reads as. - A **`human` review cannot join a group** — it suspends the whole run, which one branch of a fan-out cannot do. One failing branch fails the run; the siblings that succeeded keep their journaled results, so a retry does not re-pay for them. ## Chaining & cadence - **`chainTo`** launches a follow-up flow on success (result ⊕ input → the child's brief); `requireConfirmation` stages it for the user to confirm. - **A chain is bounded.** `chainTo` used to carry one guard — a flow could not chain to itself — so `A → B → A`, or a chain that simply ran deep, was unbounded: every hop starts a child run with a fresh idempotency key, and nothing was counting. A run now carries the chain that led to it, and a chain is refused when the target is already in that path (a cycle) or when the depth reaches `maxChainDepth` (default 5; `aiFlowsPlugin({ maxChainDepth })` or `VOLTRO_AI_FLOW_MAX_CHAIN_DEPTH`). The refusal lands on the run row's `chainRefusal` naming the path — a chain that silently does not fire is indistinguishable from one nobody declared. The parent run still **succeeds**: a refused follow-up is a configuration problem, not a reason to destroy a completed result. - **`cadence`** schedules runs — `weekly` (with `intervalWeeks` + `anchorDate`) or `monthly` (`weekOfMonth`, incl. `'last'`). `cadenceMatches` / `nextRuns` are exported so the editor's "next 3 runs" preview and the scheduler agree exactly. ## Live timeline `_voltro_ai_flow_runs` is reactive (the default): the engine patches the row (`steps[]`, `status`, `costMicroUsd`, …) as it runs, and the client's subscription streams the timeline over CDC — no polling, no sockets. ## Driving a flow from the UI `@voltro/plugin-ai-flows/web` is the browser-safe half: launch → observe → respond, without hand-rolling a subscription. ```tsx import { useFlowReview, useLaunchFlow } from '@voltro/plugin-ai-flows/web' export function ArticleFlow({ runId }: { runId?: string }) { const { launch, pending } = useLaunchFlow() const { steps, status, awaitingReview, prompt, approve, reject, done } = useFlowReview(runId) if (runId === undefined) { return ( ) } return (
    {steps.map((s) =>
  1. {s.title ?? s.type} — {s.status}
    2. )}
{awaitingReview && (

{prompt}

)} {done &&

Finished: {status}

}
) } ``` The full set: `useLaunchFlow`, `useFlowRun` (the reactive run row projected into a timeline — steps, status, the pending review, `done`), `useFlowRuns`, `useFlows`, `useRetryFlow`, `useCancelFlow`, `useRespondToFlow` (`approve` / `reject` / `choose` / `submitText`) and `useFlowReview`, which is the whole review widget in one call. **Every hook takes the tag set.** This plugin ships no fixed RPC routes — its procedures are helpers you wire into your own thin rpc files (below) — so the hooks default to `aiFlows.launch` / `.run` / `.respond` / … and accept an override plus an `apiName`: ```tsx import { useFlowRun } from '@voltro/plugin-ai-flows/web' const run = useFlowRun(runId, { apiName: 'admin', tags: { run: 'flows.oneRun' } }) ``` ## Wiring (once per app) The engine runs as one durable workflow; the operations are server helpers (a plugin route's context is subject-only, so the app owns the thin RPC surface): ```ts // api/flows/flow.run.workflow.tsx export { flowRunWorkflow as default } from '@voltro/plugin-ai-flows/workflow' // api/flows/flow.run.workflow.server.tsx import { buildFlowRunExecute, makeMediaGenerator } from '@voltro/plugin-ai-flows' export default (ctx) => buildFlowRunExecute(ctx, { resolveAgent: (ref) => /* → { system, model } */, // No AI Gateway key? Route model strings to a direct provider (or honour a // `provider/model` prefix). Absent → every call goes to the gateway. resolveModel: (model) => providerFromEnv(), // `run.tenantId` is the durable tenant from the run row — pin persistence to // it so a resume stores into the right tenant even when the subject has none. generateMedia: makeMediaGenerator({ put: ({ data, mediaType, run }) => storage.put({ bytes: data, contentType: mediaType, tenantId: run.tenantId }), ingestUrl: (u, run) => storage.ingestUrl(u, { tenantId: run.tenantId }) }, { resolveModel: (m) => providerFromEnv() }), onEvent: (e) => /* notify via @voltro/plugin-notifications */, memoryPrefix: (ownerId) => /* long-term memory */, }) // api/flows/aiFlows.launch.action.server.ts (+ retry / cancel / respond) import { launchFlow } from '@voltro/plugin-ai-flows' export default (ctx) => (input) => launchFlow(ctx, input) // The plugin OWNS _voltro_ai_flows — import its CRUD helpers over the plugin's table // instead of forking your own entity + a store bridge: import { getFlow, listFlows, createFlow, updateFlow, deleteFlow } from '@voltro/plugin-ai-flows' // e.g. api/flows/aiFlows.list.query.server.ts → (ctx) => () => listFlows(ctx) // api/flows/flows.cadence.cron.tsx import { defineSchedule } from '@voltro/runtime' import { runCadenceTick } from '@voltro/plugin-ai-flows' import { Effect } from 'effect' export default defineSchedule({ cron: '*/15 * * * *', timezone: 'Europe/Berlin', handler: (s) => Effect.promise(() => runCadenceTick(s.app)) }) ``` Adopt `@voltro/plugin-row-history` on `_voltro_ai_flows` for automatic edit history. ## Deployment notes - **Durability**: each generation is a durable step — deterministic runs resume from the first unfinished step; agentic runs aren't replay-deterministic (LLM planner), so a mid-run crash re-plans (`retry` = fresh run). - **Cadence + scale-to-zero**: `dormancy: 'sleep'` won't fire an in-process tick — use an external waker or `trigger: 'external'`. - **Run retention**: `_voltro_ai_flow_runs` is bounded by the framework sweep at **90 days** (`runsTtlMs` / `VOLTRO_AI_FLOW_RUNS_TTL_HOURS`). Only **terminal** runs are swept (`succeeded | failed | cancelled`) — a run parked on a human review is live state, not history, and a plain time-TTL would delete pending approvals. `_voltro_ai_flows` (the DEFINITIONS) is deliberately unbounded: its size tracks how many flows a team writes, not traffic. **Upgrading a live app:** the first sweep runs ~30 s after boot and deletes every terminal run older than the TTL, in batches, until the backlog drains. An app that has been running flows for more than 90 days loses that history at once — set the env var (or `runsTtlMs`) BEFORE deploying if you need it. An app's own `registerRetention` for the table also outranks the plugin's. - **Media retention**: not automatic — add a `*.cron.tsx` deleting old artifacts via `storage.delete(id)` (a plain retention sweep would orphan the blobs). Note this interacts with the run TTL: a run's steps carry hosted URLs whose blobs belong to the storage plugin, so deleting the row orphans them. Keep the run TTL at or above your media-purge window, or purge by run id first. - **Inspect page size**: `/flows` and `/runs` take a `?take=`, clamped by `inspectPageMax` (default 200). It matters here more than usual — a run row carries every step's full text output. - **Media tenant on resume**: `generateMedia` receives `run.tenantId`, read from the durable run row. Pin persistence to it (as above) rather than the caller subject — a resumed run executes under a tenant-less system subject, so reading the tenant from the subject would fail closed or store into the wrong tenant. - **Cost** is tracked in `costMicroUsd` (micro-USD) on the run + per step; project to your display currency in the view.