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

// app.config.ts
import { aiFlowsPlugin } from '@voltro/plugin-ai-flows'
export default { plugins: [aiFlowsPlugin()] }   // contributes _voltro_ai_flows + _voltro_ai_flow_runs

The contributed tables use tenant(), audit() and softDelete(). Your app must declare its own tenants table (for example in database/tenants.entity.ts) before SQL schema migration; the CLI supplies the built-in actors table. Launching a flow requires an authenticated subject with the intended tenant and an existing tenant row. The subject’s actor ID must also exist in actors for the audit foreign keys. Installing this plugin does not provision tenants or configure authentication. An in-memory inspect fixture alone does not validate these SQL and request-context prerequisites.

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

Text generation parameters

Deterministic text, agent and structured steps accept the same FlowTextParams. Code-defined steps and stored _voltro_ai_flows.steps use the same decoder and forward the values to the AI generation call:

import { flowStep } from '@voltro/plugin-ai-flows'

flowStep.text({
  prompt: 'Summarize {{document}}',
  params: {
    temperature: 0,
    maxTokens: 1200,
    providerOptions: { gateway: { order: ['openai'] } },
  },
  outputKey: 'summary',
})

maxTokens must be a positive safe integer; temperature a finite nonnegative number. Omitted options stay omitted, and zero temperature stays zero. providerOptions is a provider-keyed JSON object; which native settings work depends on the selected model/provider. Unknown top-level fields and invalid values fail the step before model or agent resolution. params cannot override prompt, system or provider: use the step fields and the agent/model resolver. Media steps have their separate media parameter contract. These are authored step settings, not a promise that an agentic planner executes that authored plan.

`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' } },
    },
  },
})

The complete authored schema is passed to generateObjectJson and validated without a partial Effect-schema translation. Local $ref, allOf, bounds and additionalProperties remain constraints, not annotations to discard. Validation does not coerce values or remove extra properties to make an invalid result pass. A missing, invalid or unsupported schema fails the step before generation; remote references and asynchronous schemas are not accepted. The selected model may impose additional limits on the schemas it can generate against.

The partial converter exports jsonSchemaToEffectSchema and JsonSchemaNode have been removed. Outside a Flow, use generateObjectJson({ jsonSchema, prompt }) from @voltro/ai; an editor preview must validate the actual JSON Schema. voltro update prints the migration for Flow users, but cannot rewrite stored database rows or infer their intended parameters and constraints. Review those definitions as well as code-defined flows.

Media inputs and artifacts

mediaInputs resolves before generation, for code and stored definitions. from: 'upload' supplies an absolute HTTP(S) URL and may declare mediaType; from: 'step' reads a brief field or earlier outputKey. A generator receives the URL map as inputs and, beside it, inputRefs — { url, mediaType } for every slot whose type is known: an upload that declared one, or a prior step's artifact, whose type the engine persisted. A model that requires typed references (inputReferences: [{ type: 'url', url, mediaType }]) reads inputRefs; a type is never guessed from a URL. The value must actually be a URL: objects are not stringified, absent/skipped outputs fail, and embedded username/password URLs are refused. Signed query strings remain unchanged. This is URL-shape validation, not authorization, a reachability check or SSRF protection: the host must allow the source and provider disclosure. Relative/browser-local URLs are not usable by a remote model. Code validation also refuses forward/same-parallel-group refs.

import { defineFlow, flowStep } from '@voltro/plugin-ai-flows'

defineFlow({
  name: 'animate-variant',
  steps: [
    flowStep.media('image', { prompt: 'Draw a landscape', params: { n: 2, seed: 0 }, outputKey: 'pictures' }),
    flowStep.media('video', {
      prompt: 'Animate the selected landscape',
      mediaInputs: {
        referenceImageUrl: { from: 'step', ref: 'pictures', artifact: 1 },
        lastFrameImageUrl: { from: 'upload', url: 'https://assets.example.com/final.png' },
      },
      params: { generateAudio: false, seed: 0 },
      outputKey: 'video',
    }),
  ],
})

Without artifact, a step reference uses its primary URL. artifact: 0 selects the first returned artifact; other nonnegative indices select that position. An absent index fails; the engine never silently substitutes the primary.

The built-in makeMediaGenerator maps these slots to native AI inputs:

Modality	Slots
image	`referenceImageUrl` → reference image; `maskImageUrl` → mask (requires the reference)
video	`referenceImageUrl` → image; `firstFrameImageUrl` / `lastFrameImageUrl` → typed frame roles; `referenceVideoUrl` → video reference
audio	No reference-input support; a supplied slot is refused

Custom generators receive resolved inputs with their own slot names. The built-in adapter refuses unknown slots and unknown/invalid params:

Modality	Parameters
image	`n`, `seed`, `size`, `aspectRatio`, `providerOptions`
video	`aspectRatio`, `durationSeconds`, `resolution`, `fps`, `seed`, `generateAudio`, `maxInlineBytes`, `providerOptions`
audio	`voice`, `outputFormat`, `instructions`, `speed`, `language`, `providerOptions`

Zero seed and explicit false survive. Counts/budgets are positive safe integers; durations/rates are finite positive numbers; sizes/ratios have positive integer components. Provider-specific support still depends on the chosen model. The agentic generate_media tool also accepts resolved inputs and params; the planner chooses them rather than executing the authored media plan verbatim.

Every artifact keeps its identity

A custom MediaGenerator returns artifacts, each with url, mediaType, role and optional storageRefId / per-artifact providerMetadata. Exactly one has role primary; other roles are host-defined (for example preview). The built-in adapter persists every returned image, marking the first primary and the rest variant. Audio/video produce one primary artifact.

RunStep.url and the result bundle's outputKey project the primary URL for prompt composition. All variants remain in RunStep.artifacts, through the workflow journal, sequential/parallel timeline and the browser codec:

import { useFlowRun } from '@voltro/plugin-ai-flows/web'

const { steps } = useFlowRun(runId)
const artifacts = steps.flatMap(step => step.artifacts ?? [])

requestId is optional provider/usage correlation, not a storage ID. For storage cleanup or renewed delivery URLs, use each artifact's storageRefId. A URL can expire; a reference ID is not a promise of permanent URL access.

Persistence callbacks return Effects in the host runtime, retaining services and interruption. Map a storage result's id to refId explicitly and obtain a delivery URL through the storage API; StorageRef has no url field. The wiring below assumes an already authorized server pipeline: getUrl is not an access check for private files. Apply authorization before sharing a URL with a model or end user; configure absolute delivery URLs for provider references.

This does not make multi-artifact storage atomic or external generation crash-safe. If a later persistence operation fails, earlier writes may exist; keep reconciliation for that case. voltro update explains the new callback and result shapes, but cannot recover variants or reference IDs discarded by an earlier run. Existing stored definitions also need review.

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. Send the displayed reviewStepIndex alongside runId and response. respondToFlow compares it with the live timeline and refuses stale clicks. It computes the actual execution ID from the stored workflow payload; an idempotency key is not an execution ID. Missing or ambiguous timeline identity refuses instead of guessing zero or currentStep.

Host RPC descriptors use FlowReviewInput and FlowReviewReceipt from @voltro/plugin-ai-flows/ir. Include FlowReviewDeliveryError from @voltro/plugin-ai-flows/errors alongside FlowRunNotFound and FlowNotAwaitingResponse in the declared error schema. All three schema imports are browser-safe; respondToFlow belongs in the server executor.

The receipt { runId, stepIndex, executionId, eventId, status: 'accepted' } acknowledges delivery, not completion of the run. Both dev and serve await signal storage and suspended-workflow completion before returning it. Missing runtime or invalid stored identity fails before signalling. delivery-unknown means the answer may already have taken effect: observe the run before retrying. The signal row and deferred completion are not one atomic transaction. The engine projects humanResponse after consuming the answer; it is not a sender attempt record or proof that every projection write succeeded.

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.

Cancellation: row state is not engine acknowledgement

cancelFlow(ctx, { runId }) returns FlowCancelReceipt (also exported from @voltro/plugin-ai-flows/ir for browser-safe action output schemas). status describes the flow row. engine separately reports cancelled, accepted, unknown, already-terminal, unreachable or not-requested. Only engine: 'cancelled' confirms engine cancellation. An accepted request or a lost acknowledgement must not produce a “stopped” success message.

The row carries the same uncertainty. A live row (pending, running, waiting) is written cancelling with cancelledAt set to the request time. It reads cancelled — with completedAt — only once the engine confirmed: the receipt reports engine: 'cancelled', the engine honours the request at its next step boundary, the engine's own run row turns cancelled (the plugin mirrors that onto the flow row), or a repeated cancelFlow gets confirmation. A reader that only sees the row later therefore cannot mistake an unheard request for a stopped run. cancelling is not terminal: useFlowRun and useFlowReview report done only for succeeded, failed or cancelled, and a cancelling run is no longer awaitingReview.

Two consequences for a UI, because cancelledAt being set on a run that is still executing is the part that surprises. An elapsed-time function written as completedAt ?? cancelledAt freezes the clock the moment cancellation is requested, while the run keeps going — read completedAt alone for the end of a run, and treat cancelledAt as when someone asked. And a status map that does not know cancelling does not fail: the label renders as its untranslated key with a default chip, so nothing points at the omission. A switch over RunStatus with no default arm is what turns that into a compile error.

What a UI shows per engine outcome:

`engine`	row `status`	Render as
`cancelled`	`cancelled`	Cancelled.
`accepted`	`cancelling`	Cancelling — the engine will honour it; observe `control.requestId` with `ctx.workflows.readControl`.
`unknown`	`cancelling`	Cancelling — acknowledgement lost; the interruption may already be in effect. Retry `cancelFlow` or inspect the execution.
`unreachable`	`cancelling`	Cancellation requested, the engine was not reached; the run may still be resident. Retry `cancelFlow` once the runtime is available.
`already-terminal`	the flow result's own status	Finished before the request: show that status.
`not-requested`	`succeeded`, `failed` or `cancelled`	The row was already terminal; nothing was requested.

control retains the complete workflow control receipt, including its request ID and execution snapshot, or is null when no valid receipt was obtained. With a non-null request ID, use ctx.workflows.readControl with that receipt's operation, requestId, snapshot.workflowName and snapshot.executionId to observe the original request without submitting another one. Without a request ID there is no resident request to observe; inspect the execution instead.

The conditional row write protects a concurrently completed flow and preserves the request timestamp on repeated cancellation. Terminal rows are never resubmitted to the engine: not-requested means the flow row already says succeeded, failed or cancelled; it is not an engine observation. unreachable means the runtime or stored execution address is unavailable. A rejected engine call is unknown, since the interruption may already have taken effect. An already-terminal engine result can reconcile a stale row; the flow result's own status wins over a successful engine envelope. Cancellation is not rollback of provider charges or other external effects.

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.

`flow.run` is internal

The flow.run workflow this plugin ships declares internal: true, so it has no start rpc: a browser cannot start it by tag. Every start is server-side — this plugin's own guarded procedures, the cadence tick, and chainTo — which is what it always was in practice. The payload carries runId, the id of a row the caller must already have created, so a wire start could only fail or patch somebody else's run.

If you re-export the descriptor from your own flow.run.workflow.tsx, that is still the supported wiring; the decision travels with it, so your app boots under security.defaultDeny without adding anything.

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, reviewStepIndex, 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 && reviewStepIndex !== undefined && (
        <div>
          <p>{prompt}</p>
          <button onClick={() => approve(runId, reviewStepIndex)}>Approve</button>
          <button onClick={() => reject(runId, reviewStepIndex)}>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.

Response callbacks take runId, stepIndex first: approve/reject then accept optional text; choose takes a value and submitText takes text. The low-level form is respond(runId, stepIndex, response), matching respondToFlow(ctx, { runId, stepIndex, response }). Index zero is valid; an undefined index must disable the form. voltro update reports the required host-schema and call-site changes; it cannot infer which review a user saw.

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)

launchFlow / retryFlow and their web hooks return { runId, requestId, admission }. Use the browser-safe FlowStartReceipt from @voltro/plugin-ai-flows/ir as your launch/retry action's output schema. Read admission.status for running, queued, dropped, skipped or replayed; observe the run row separately for execution progress. A retry keeps its prior failure until the executor changes it, even when submission is accepted. The observed admission is stored in workflowStart. A write failure after submission propagates and does not prove that nothing started. voltro update reports the output-schema and consumer changes; regenerate and rebuild the client after updating your descriptors.

The run row and its start are one fact: the row (status: 'pending') and the workflow's durable arrival commit in the same transaction, so a crash leaves either both or neither. What that arrival BECOMES is decided later by the workflow layer, and the row follows it two ways. An admitted start is caught live — the engine's run row appearing under a queued handle turns the handle into the accepted execution. A start that will never run (dropped by a rate limit, skipped for an incumbent, abandoned after timeouts.start or a dead-letter) leaves no engine row behind, so reconcileFlowStarts(ctx) reads every pending row's receipt through workflows.readStart and marks the row failed with error: 'FlowStartNotExecuted' and the reason. The cadence tick runs that reconciliation first on every replica; a read failure refuses the tick, a per-row failure is carried in its report. Call it yourself from any server context that holds an AppContext when you do not run the cadence.

launchFlow and retryFlow require ctx.workflows. Without it they refuse before writing run state: launch creates no pending row, and retry preserves the existing failure. Missing runtime is a host wiring defect, not a successful queue receipt. This preflight does not make the run-row write and workflow submission atomic; pending alone is not proof of confirmed engine acceptance. launchFlow and retryFlow open ctx.transaction themselves, including from an action or schedule. The run row, workflowStart and durable start intent share that transaction. Inside a mutation it joins the existing transaction. Its admission is queued with deferral.mode: 'transaction', provisional until the owning commit. Rollback removes all three; recovery can find the committed intent without a post-commit callback. Retry reads and validates the persisted payload inside that transaction, retaining FlowRunNotFound as a typed failure. Cadence opens the same boundary per candidate. It is not proof of regional worker discovery or completed execution.

Retry also requires a stored, non-empty string requestId. A missing or malformed identity is refused without changing the previous failure or submitting work. Retry never invents a new identity for an existing run. If the original identity cannot be recovered, an explicitly new launch is a new run, not a continuation of the old workflow journal.

The stored payload must also satisfy the workflow schema: a non-empty flowRef, an object input, and source: 'manual' | 'cron' are required. Retry does not replace missing input with {} or a missing source with 'manual'. Invalid stored payloads fail before changing the run; the diagnostic does not print their contents. Valid inputs and the source are forwarded unchanged.

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'
import { StorageService } from '@voltro/plugin-storage'
import { Effect, Schema } from 'effect'
export default (ctx) => buildFlowRunExecute(ctx, {
  resolveAgent:  (ref, run) => resolveTenantAgent(run.tenantId, ref), // → Effect<{ system, model }, FlowNotFound>
  // 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 }) => Effect.gen(function* () {
      const storage = yield* StorageService
      const bytes = typeof data === 'string' ? yield* Schema.decodeUnknown(Schema.Uint8ArrayFromBase64)(data) : data
      const ref = yield* storage.put({ bytes, contentType: mediaType, tenantId: run.tenantId })
      const url = yield* storage.getUrl(ref.id, { tenantId: run.tenantId })
      return { url, refId: ref.id }
    }),
    ingestUrl: (source, run) => Effect.gen(function* () {
      const storage = yield* StorageService
      const ref = yield* storage.ingestUrl(source, { tenantId: run.tenantId })
      const url = yield* storage.getUrl(ref.id, { tenantId: run.tenantId })
      return { url, refId: ref.id }
    }),
  }, { 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)) })

Current scheduler limitation: the wiring example above receives the scheduler's system context (tenantId: null). Stored flows carry their row tenant into each candidate's transaction. Code flows have no row owner and still require an explicitly selected tenant; otherwise their run write refuses with TenantScopeViolation. Discovery across physical databases remains unfinished. Do not work around this by spreading a context and changing request.subject: its store and workflow facade remain bound to the original identity. Full scheduler tenant placement remains unfinished.

For an explicitly selected tenant, use the complete transaction context:

await s.app.transaction(tx => runCadenceTick(tx, s.scheduledAt), { tenantId })

This path is tested through the real scheduler for code and stored flows: the run and queued receipt carry the selected tenant, and system audit attribution stays null. The enclosing transaction rolls back the whole tick on failure. It does not discover tenants or physical databases; choose the tenant explicitly and do not interpret a successful schedule as completed Flow execution.

With residency or namespace isolation enabled, a tenantless system context starts unplaced. Select transaction(work, { tenantId }) or storeForTenant(tenantId) before using database-backed capabilities; direct store, loader, workflow, event, outbox, connection and webhook access refuses until selection. There is no primary-store fallback. Physical tenant discovery and process-restart verification remain separate open work.

cronRequestId(flowRef, now, tenantId) returns an opaque identifier for the tenant, flow and absolute minute. Do not parse it. Repeated ticks in that minute reuse the ID; another tenant gets a different one. Deploy the ID change between cadence slots: submissions from different builds within one slot are not deduplicated.

runCadenceTick counts only a running admission receipt in started, not queued, dropped, skipped or replayed. A candidate-read failure rejects the tick instead of reporting an empty successful scan. Per-flow persistence or submission failures are collected in an AggregateError after attempting the remaining candidates; inspect its errors to diagnose the failures. Missing workflow runtime refuses before writing a pending run. A conflicting run insert is resolved by the store, not inferred from an arbitrary exception; resubmission uses the scoped, validated persisted input rather than new defaults. Launch, retry and the cadence tick store their last observed admission handle in the run row's workflowStart, including the queued intent ID. This is an observation, not the current execution status; recording it never overwrites executor progress. A null field means no answer was recorded, not proof that submission failed. A receipt-write failure rejects the tick even if submission already succeeded. Each candidate uses ctx.transaction: its run write, start intent and receipt commit together. In the standard context this yields queued, so started stays empty until a genuine running admission is observed; matched is not an execution count. An enclosing transaction still owns the commit. This does not prove physical worker discovery or completed engine execution.

Adopt @voltro/plugin-row-history on _voltro_ai_flows for automatic edit history.

Agent resolution belongs to the run's tenant

resolveAgent(ref, run) receives { runId, tenantId, ownerId } (a FlowRunContext) from the durable _voltro_ai_flow_runs row, both for deterministic agent steps and the agentic run_agent tool. Refuse with FlowAgentUnavailable (not-found, access-revoked, model-unusable) for anything that is not "no such agent": its message is for the host — the step's error and the run's log — and only its publicMessage reaches the planner's model, which reads a resolver's words as tool output. Without a publicMessage the model sees a generic line. Use run.tenantId to scope persona lookups; use run.runId for correlation. The current request subject is not authoritative here: a system/crash resume may carry no tenant at all. The same stored tenant reaches media persistence. An explicitly stored null means a single-tenant run.

The engine refuses execution if the run row cannot be read, is absent or does not carry a valid explicit tenant. It does not substitute null. If the run row is writable, its status becomes failed with flow run context unavailable in errorMessage; no model, agent resolver or media generator is invoked. Restore the run's stored context before retrying. A database outage may also prevent writing that diagnostic; it still does not permit generation.

Run identity is not authorization. The resolver must enforce which personas the run may use; an agentic model's requested ref is not proof of access.

Resolved tools: planner and delegated agents

An AgentResolution may return tools: Record<string, AnyTool> and a positive integer maxSteps (default 8). Both deterministic agent steps and the agentic run_agent delegate pass these to the existing AI tool loop. The persona owns its tools; planner tools are not automatically granted to delegates.

EngineDeps.resolveTools(flowRef, run, planner) resolves additional agentic planner tools. It receives the original code name or stored row ID, not a display name, the durable { runId, tenantId, ownerId }, and planner.provider() — the provider the planner will run on, resolved only when asked — so a host that offers a provider-native search (web search on OpenAI, Google, Anthropic) hands over the matching tool. It runs inside the planner's journaled activity, before model I/O. A typed resolver failure fails the run: the row keeps the generic flow tool resolution failed (a run row is client-readable) and the resolver's own message goes to the server log; names run_agent, generate_media, and pick_structured are reserved and collisions refuse rather than replacing an internal tool. Without this callback only the built-in tools are available. Deterministic text/structured/media steps do not invoke it; use an agent step when tool-assisted text work is needed.

import { defineTool } from '@voltro/ai'
import type { EngineDeps } from '@voltro/plugin-ai-flows'
import { Effect, Schema } from 'effect'

const deps: EngineDeps = {
  resolveTools: (_flowRef, run, _planner) => Effect.succeed({
    run_context: defineTool({
      name: 'run_context', description: 'Read this flow run identity',
      input: Schema.Struct({}),
      execute: () => Effect.succeed({ runId: run.runId, tenantId: run.tenantId }),
    }),
  }),
}

The example only exposes run identity. For real reads/writes, authorize in the resolver and recheck at invocation where permissions may change. Bind the tools to that run's scope; no current user subject or unrestricted store is invented. Resolver and tool Effects inherit host services and cancellation. Tool functions are server-only and never serialized into Flow IR. A completed activity replays its result, not tool functions; a crash during the activity can repeat tool I/O. Tools with external side effects still need durable idempotency. This addition does not make an agentic plan deterministic or every tool call a durable step.

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. Every step also says where its number came from — costSource is 'gateway' (the provider's actual charge), 'estimated' (the price table) or 'unpriced' (the 0 beside it is not a price: an unknown model, a media call the adapter could not price). The run row folds those into costCompleteness: 'actual' when every priced step was gateway-priced, 'estimated' when none was unpriced, 'incomplete' when any was, 'unknown' before the first priced step. Bill on 'actual'; show the rest as what they are.

The dashboard lists recent run summaries without loading step outputs. Select a run to inspect its input, output, recorded steps, workflow admission and errors. The inspect endpoint accepts ?runId= to return that run alone; ?take= bounds the newest-first summary list. Missing runs are reported as unavailable, not as a successful empty result.

When the recorded workflow admission contains an execution ID, Open workflow run opens the corresponding filtered workflow view and preserves the selected replicas. Queued starts without an execution ID do not offer a run link.

The dashboard keeps the selected run ID and Close details action visible while run details are loading or unavailable. You can return to the run list without waiting for a successful detail response.

Replica-scoped inspection

Flow definitions and run reads declare process scope because a bound in-memory store can differ between replicas. The dashboard preserves the answering replica for each list and selected run detail. With multiple replicas selected, it compares their observations rather than merging counts or choosing one payload silently. Summary lists omit input, output and step payloads; selecting a run fetches its details from the same replica.

Code and stored definition catalog

The /flows inspect catalog includes code flows registered in the answering process as well as stored definitions. Code flows come first, sorted by name; stored definitions follow newest first. take and inspectPageMax bound the combined result. When a code name equals a stored ID, only the code definition appears, matching the executor's resolution precedence. Equal display names with different references remain separate definitions.

The dashboard’s Source column labels definitions as Code or Database. Older endpoints without source metadata show a dash. Each inspect row identifies definitionSource as code or stored. Code rows use the registered name as their ID and omit stored lifecycle fields and usage counts; these missing fields do not mean disabled or zero runs. Import the defining module in the app process to register a code flow. The catalog does not execute flows. Code entries contain summary fields only; they do not include prompts.

Open details when a replica leaves

In both single-replica and all-replicas views, an opened flow detail remains attached to its original replica even when that process leaves. The panel identifies the unavailable source and keeps the selected run ID and close action visible. Cached detail payloads are hidden while that source is unavailable. The selection is not silently moved to a replacement process: open the run in an available replica’s panel to inspect its current result, then close the old detail. If several plugin instances have details open on the departed replica, the observation remains until the last one closes. Changing apps clears this retained state; explicitly changing replica selection follows the new selection.

AI Flows

#Install

#Author a flow (code-first)

#Step kinds

#Text generation parameters

#structured — the schema is real

#Media inputs and artifacts

#Every artifact keeps its identity

#Human-in-the-loop

#The park is bounded, and the bound is yours

#Cancellation: row state is not engine acknowledgement

#Conditional steps and fan-out

#when: — run a step only sometimes

#group: — run steps concurrently

#flow.run is internal

#Chaining & cadence

#Live timeline

#Driving a flow from the UI

#Wiring (once per app)

#Agent resolution belongs to the run's tenant

#Resolved tools: planner and delegated agents

#Deployment notes

#Replica-scoped inspection

#Code and stored definition catalog

#Open details when a replica leaves