AI Flows

@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

// 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:

// 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)

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:

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):

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

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 outputKeys). 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.

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 (
      <button disabled={pending} onClick={() => launch({ flowRef: 'blog-article', input: { topic: 'Effect' } })}>
        Start
      </button>
    )
  }
  return (
    <div>
      <ol>{steps.map((s) => <li key={s.id}>{s.title ?? s.type} — {s.status}</li>)}</ol>
      {awaitingReview && (
        <div>
          <p>{prompt}</p>
          <button onClick={() => approve(runId)}>Approve</button>
          <button onClick={() => reject(runId)}>Reject</button>
        </div>
      )}
      {done && <p>Finished: {status}</p>}
    </div>
  )
}

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:

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):

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