# Run a durable flow

> Use a flow when the steps, values and replay history matter. Start with a small sequential graph, not an arbitrary execution engine.

- Package: `@db3.ai/app/flows`
- Canonical page: [https://db3.ai/docs/flows](https://db3.ai/docs/flows)
- Markdown: [https://db3.ai/docs/flows.md](https://db3.ai/docs/flows.md)
- Framework source of truth: `packages/app/src/flows/README.md`

<a id="setup"></a>

## Wire the service into your app

Use the independent app/tools and test SQL credentials from Installation. Flows is an explicit app extension: the base `App` does not automatically provide a `flows` getter. The example subclass registers it through the normal `service()` cache.

Every producer and worker needs this boot wiring, definition store and block registry. Initializing `Flows` registers `FlowStepJob`; that queued job resolves `app().flows` when restored. Production apps migrate `QueuedJob`, `FailedJob` and `FLOW_MODELS` before running workers.

- [Installation and test SQL](https://db3.ai/docs/installation.md#database-labs)
- [Application services](https://db3.ai/docs/app.md)

### examples/FlowExampleApp.ts

```typescript
import { FileFlowDefinitionStore, Flows } from '@db3.ai/app/flows';
import { App, type AppOptions } from '@db3.ai/app/server';
import { normalizeTextBlock } from './normalizeTextBlock';
import { uppercaseTextBlock } from './uppercaseTextBlock';

/** Application-owned boot wiring shared by request producers and queue workers. */
export class FlowExampleApp extends App {
	readonly #definitionRoot: string;
	/** Creates the app; the caller owns its definition directory and database. */
	constructor(options: AppOptions, definitionRoot: string) { super(options); this.#definitionRoot = definitionRoot; }
	/** Lazily registers the durable flow-step handler and the app's block types. */
	get flows(): Flows {
		return this.service('flows', () => new Flows({ queue: this.queue, queueName: 'flows', maxTries: 1, definitions: new FileFlowDefinitionStore({ root: this.#definitionRoot }), blocks: [normalizeTextBlock, uppercaseTextBlock] }));
	}
}
```

<a id="copy"></a>

## Copy the flow

The five files separate app wiring, graph data, two executable blocks and the disposable runner. Definitions are stored in temporary files by this lab; keep application definitions in a deliberate versioned location.

### Copy the shipped example

```bash
mkdir -p examples
cp -R node_modules/@db3.ai/app/src/flows/examples/. examples/
```

<a id="run"></a>

## Run and inspect it

Expect `completed`, output `CLIENT NOTE`, two steps and a recorded progress log. Blank input fails with `Note text cannot be blank.`. Replaying repaired input with the original definition gives `FIXED NOTE`; replaying the latest definition gives `Updated: CLIENT NOTE`. All temporary tables and files are removed.

### Run the lab

```bash
npx tsx examples/runTextFlow.ts
```

<a id="block"></a>

## Give each block a small contract

`defineBlock()` describes named input/output ports, optional configuration and one executable function. The graph validates basic value types. Your block still validates business meaning, such as a string that contains more than whitespace.

Keep database work in app-owned blocks with scoped models. There is no generic arbitrary-SQL block. The context gives run/step identities and durable `log()`; avoid putting secrets or sensitive source values in it.

### examples/normalizeTextBlock.ts

```typescript
import { defineBlock } from '@db3.ai/app/flows';

/** Validates application meaning after the graph's basic string contract. */
export const normalizeTextBlock = defineBlock<{ text: string }, { text: string }>({
	type: 'notes.normalize', name: 'Normalize note',
	inputs: { text: { type: 'string', required: true } },
	outputs: { text: { type: 'string', required: true } },
	/** Rejects blank text and records a safe, value-free progress message. */
	async run(input, context) {
		const text = input.text.trim();
		if (!text) throw new Error('Note text cannot be blank.');
		await context.log('info', 'Note normalized');
		return { text };
	},
});
```

<a id="configuration"></a>

## Separate configuration from input

The formatting block has a `prefix` configuration default. Configuration is resolved from the block occurrence, while text arrives through a named connection. Editor hints can help a host render a form; they do not change runtime validation.

### examples/uppercaseTextBlock.ts

```typescript
import { defineBlock } from '@db3.ai/app/flows';

/** Formats validated text without an external provider or irreversible side effect. */
export const uppercaseTextBlock = defineBlock<{ text: string }, { text: string }, { prefix: string }>({
	type: 'notes.uppercase', name: 'Uppercase note',
	inputs: { text: { type: 'string', required: true } },
	outputs: { text: { type: 'string', required: true } },
	config: { prefix: { type: 'string', default: '' } },
	/** Applies the snapshotted configuration to one deterministic output. */
	run(input, context) { return { text: `${context.config.prefix}${input.text.toUpperCase()}` }; },
});
```

<a id="definition"></a>

## Keep the graph as source data

The definition contains stable ULIDs, ports, connections and positions. Preserve IDs when editing an existing graph. `saveDefinition()` validates before writing; pass `expectedRevision` to detect a stale file-store edit.

The current compiler accepts one connected sequential path. Branches, merges and cycles are rejected. A visual position does not define execution order. The file store writes deterministic JSON and uses content revisions; it is not a multi-writer database transaction or untrusted filesystem sandbox.

### examples/createTextFlow.ts

```typescript
import type { FlowDefinition } from '@db3.ai/app/flows';
import { ulid } from '@db3.ai/pure/ulid';

/** Creates a two-block sequential definition with stable IDs retained on edits. */
export function createTextFlow(): FlowDefinition {
	const normalize = ulid();
	const uppercase = ulid();
	return {
		schemaVersion: 1, id: ulid(), name: 'Normalize note text',
		inputs: { text: { type: 'string', required: true } }, outputs: { text: { type: 'string', required: true } },
		blocks: [{ id: normalize, type: 'notes.normalize', position: { x: 0, y: 0 } }, { id: uppercase, type: 'notes.uppercase', config: { prefix: '' }, position: { x: 250, y: 0 } }],
		connections: [{ id: ulid(), sourceBlockId: normalize, sourcePort: 'text', targetBlockId: uppercase, targetPort: 'text' }],
	};
}
```

<a id="worker"></a>

## Dispatch and process the steps

`run()` creates durable run/step state and dispatches a flow-step job to `flows`. The lab drains ready jobs with a finite safety limit. Production needs a supervised queue worker consuming that same queue.

`runDetails()` returns the run, ordered steps and lifecycle/log events. Captured input/output has a default one-megabyte boundary. There is no automatic secret redaction or tenant authorization; scope any history/API access in your app.

### examples/runTextFlow.ts

```typescript
import { mkdtemp, rm } from 'node:fs/promises';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { pathToFileURL } from 'node:url';
import { createGeneratedTestDatabase } from '@db3.ai/app/db/test/db';
import { FLOW_MODELS } from '@db3.ai/app/flows';
import { FailedJob, QueuedJob } from '@db3.ai/app/queue';
import { FlowExampleApp } from './FlowExampleApp';
import { createTextFlow } from './createTextFlow';

/** Drains only ready jobs, with an explicit finite guard for this small lab. */
async function drain(application: FlowExampleApp): Promise<void> {
	for (let jobs = 0; jobs < 20; jobs++) if (!await application.queue.processNextJob('flows')) return;
	throw new Error('Flow lab exceeded its 20-job safety limit.');
}

/** Runs a real durable graph, rejects invalid input and replays original/latest definitions. */
export async function runTextFlow() {
	const root = await mkdtemp(join(tmpdir(), 'db3-flow-guide-'));
	try {
		const database = await createGeneratedTestDatabase('flow_guide');
		const application = new FlowExampleApp({ db: database.db, queue: { queue: 'flows', retryDelaySeconds: 0, queueMonitor: false } }, root);
		try {
			await application.db.install(QueuedJob, FailedJob, ...FLOW_MODELS);
			const flow = createTextFlow();
			const stored = await application.flows.saveDefinition(flow);
			const original = await application.flows.run(flow.id, { text: '  Client note  ' });
			await drain(application);
			const originalDetails = await application.flows.runDetails(original.id!);
			const failed = await application.flows.run(flow.id, { text: '   ' });
			await drain(application);
			const failure = await application.flows.runDetails(failed.id!);
			const changed = structuredClone(flow);
			changed.blocks[1]!.config = { prefix: 'Updated: ' };
			await application.flows.saveDefinition(changed, { expectedRevision: stored.revision });
			const replay = await application.flows.replay(failed.id!, { definition: 'original', input: { text: 'Fixed note' } });
			const latest = await application.flows.replay(original.id!, { definition: 'latest' });
			await drain(application);
			const replayed = await application.flows.runDetails(replay.id!);
			const latestDetails = await application.flows.runDetails(latest.id!);
			return { status: originalDetails.run.status, output: originalDetails.run.output, steps: originalDetails.steps.length, logged: originalDetails.events.some(event => event.message === 'Note normalized'), failure: failure.run.status, failureMessage: failure.run.error?.message, repaired: replayed.run.output, latest: latestDetails.run.output, replayLinked: replayed.run.replayOf?.id === failed.id };
		} finally { try { await application.close(); } finally { await database.destroy(); } }
	} finally { await rm(root, { recursive: true, force: true }); }
}

if (process.argv[1] && import.meta.url === pathToFileURL(process.argv[1]).href) console.log(JSON.stringify(await runTextFlow(), null, 2));
```

<a id="replay"></a>

## Repair, then choose what to replay

Replay creates a new linked run. `definition: "original"` uses the stored graph snapshot; `"latest"` resolves current source. You can supply replacement input explicitly. The lab changes the second block configuration and proves the two modes differ.

The snapshot contains definitions, not a copy of executable JavaScript. Replaying an old graph still uses registered runtime code. Version block types when code changes would break historic behavior. Side-effecting blocks need idempotency: replay can repeat work.

<a id="nested"></a>

## Compose a child flow

Structural `flow.input`, `flow.output` and `flow.subflow` blocks are registered by the service. A subflow occurrence supplies a child `flowId`; its public ports come from the child definition.

The parent waits while a linked child run executes. Nested definitions are captured for original replay; latest replay resolves current source. The service integration tests cover nested waiting/failure/replay, but the first lab stays a two-block path. A nested-flow cookbook and designer integration remain follow-ups.

<a id="testing"></a>

## Testing

Create a `tests/flows` directory and save the test below as `runTextFlow.test.ts`. It imports the example you copied into `examples/`. Keep the same folder layout so the relative import resolves.

Run from the application root with the development dependencies from Installation. These are consumer tests, not commands that assume a framework checkout. This database lab needs the same test-only SQL credentials when run through Vitest.

### tests/flows/runTextFlow.test.ts

```typescript
import { expect, it } from 'vitest';
import { runTextFlow } from '../../examples/runTextFlow';

it('runs a persisted graph, captures safe progress and repairs failed work through explicit replay', async () => {
	expect(await runTextFlow()).toEqual({ status: 'completed', output: { text: 'CLIENT NOTE' }, steps: 2, logged: true, failure: 'failed', failureMessage: 'Note text cannot be blank.', repaired: { text: 'FIXED NOTE' }, latest: { text: 'Updated: CLIENT NOTE' }, replayLinked: true });
});
```

<a id="run-tests"></a>

## Run and extend the test

Change the prefix or add another pure block. Keep the failure, repair, replay-link and original/latest assertions. This lab uses one attempt to make terminal failure immediate; configure retry policy for the actual operation rather than copying that as a production default.

### Run your copied test and check types

```bash
npx vitest run tests/flows/runTextFlow.test.ts
npx tsc --noEmit --target ES2022 --module ESNext --moduleResolution Bundler --types node --skipLibCheck examples/*.ts
```

<a id="coverage"></a>

## Coverage and API

Tested: real file definitions and SQL queue, App boot, two blocks, input/business validation, progress, terminal failure and replay modes. Explained/reference: nested graphs, editor contracts, payload bounds, revisions and production safety. Branching, generic SQL execution, credentials in captured values and designer UI are not promised.

- [Complete Flows API](https://db3.ai/docs/flows-api.md)
- [Queue operation](https://db3.ai/docs/queue-overview.md)

## Behavioural verification
Persists and runs a two-block graph, inspects steps/logs, fails blank input and replays original/latest definitions.
- Behaviour test: `packages/app/src/flows/tests/examples/runTextFlow.test.ts`
- Repository test command (framework checkout only): `npm run test:service --workspace @platform/app -- flows --maxWorkers=1`
- In an installed application, use the walkthrough commands instead of this repository test.
- Expected outcome: The guide test passes against the real framework components.
- Environment: Disposable SQL queue, temporary file definition store, no provider or graph UI.

## Related documentation
- [Queue](https://db3.ai/docs/queue-overview.md): Create durable background jobs, run named workers, understand every attempt, compose chains and batches, and recover failures through one complete service guide.
- [Scheduler](https://db3.ai/docs/scheduler.md): Decide when a daily task is due, claim its occurrence and hand expensive work to Queue. Inspect what happened without hiding schedule definitions in a database.
- [Restore durable application objects](https://db3.ai/docs/serialization.md): Save the state needed to call one registered constructor again. Keep services and connections out of the payload.

## Framework-owned source: `packages/app/src/flows/README.md`

This is the exact source document captured by the documentation build. Use it for detailed API and workflow guidance, subject to the public package exports and behavioural evidence identified above.

````markdown
# Flows

## Runnable flow and testing

The base `App` does not automatically supply a `flows` property. Copy the
installed package's `src/flows/examples/` into `examples/` to use the explicit
`FlowExampleApp` service extension, then run `npx tsx examples/runTextFlow.ts`
with disposable SQL credentials. It creates/removes its own test database and
file-definition directory. Two blocks normalize and format text; the lab checks
steps, progress, blank-input failure and original/latest replay after repair.
The website includes the exact `tests/examples/runTextFlow.test.ts` consumer
test. A definition snapshot preserves graph/configuration, not executable code;
version block types when changing runtime behavior for historical graphs.

Flows are durable, observable graphs layered over the existing queue. A flow definition is the source of truth for both the runtime and visual designer. It contains stable ULIDs, named input/output contracts, configured block occurrences, named-port connections, and graph positions.

Definitions are intentionally separate from run data:

- A `FlowDefinitionStore` reads and writes source definitions.
- `FileFlowDefinitionStore` stores deterministic `flow.json` files suitable for Git.
- `FlowRun`, `FlowStepRun`, and `FlowRunEvent` persist invocations, every block boundary, logs, failures, timing, and replay lineage.
- `FlowStepJob` executes one block at a time through the application's existing queue.

The initial compiler accepts one connected sequential path. Branches and merges are rejected until their scheduling and replay semantics are explicit.

## App Service

An app exposes its configured service as `app().flows`:

```ts
import { fileURLToPath } from 'node:url';
import { FileFlowDefinitionStore, Flows } from '@db3.ai/app/flows';

class App extends FrameworkApp {
	get flows(): Flows {
		return this.service('flows', () => new Flows({
			queue: this.queue,
			queueName: 'flows',
			definitions: new FileFlowDefinitionStore({
				root: fileURLToPath(new URL('./flows', import.meta.url)),
			}),
			blocks: flowBlocks,
		}));
	}
}
```

`FlowStepJob` resolves that service from the active application when the queue
worker handles it. No flow-specific queue context or service injection is
required.

Install `QueuedJob`, `FailedJob`, and `...FLOW_MODELS` in the app database schema.

## Function Blocks

A function-backed block is one file with a serializable contract and one function:

```ts
import { defineBlock, type FlowValues } from '@db3.ai/app/flows';

type TextValues = FlowValues & {
	text: string;
};

export default defineBlock<TextValues, TextValues>({
	type: 'text.uppercase',
	name: 'Uppercase',
	inputs: {
		text: { type: 'string', required: true },
	},
	outputs: {
		text: { type: 'string', required: true },
	},
	run: input => ({
		text: input.text.toUpperCase(),
	}),
});
```

The block context exposes durable `log(level, message, data)` entries. A debug tap can therefore log input and return it unchanged without introducing a separate instrumentation system.

Database operations should be explicit app-owned block types such as `database.fetch-content-plan` or `database.insert-article`. These blocks use the app's models and ownership rules. The framework deliberately does not expose a generic arbitrary-table or SQL block, because that would bypass model conversion, tenancy, authorization, and reviewable contracts.

## Definition-Driven Inputs

Public flow inputs and block configuration use `FlowValueDefinition`. In addition to runtime type, requirement, description, and default information, a value may carry optional host-agnostic editor hints:

```ts
const request = {
	type: 'json',
	required: true,
	editor: {
		component: 'DomJsonInput',
		label: 'Article request',
		rows: 8,
	},
} satisfies FlowValueDefinition;
```

Editor metadata never changes runtime validation. A host may map component names to DOM form fields, native controls, or another component registry while keeping the flow definition as the source of truth.

## Flow Boundaries And Subflows

Every `Flows` service registers three framework-owned structural blocks:

- `flow.input` exposes the containing definition's public inputs inside the graph.
- `flow.output` terminates the graph and exposes the containing definition's public outputs.
- `flow.subflow` groups another definition behind one generic nested-flow block.

The child definition is the only public-contract source for a subflow. A parent occurrence only needs the generic type and the child ULID:

```json
{
	"id": "01KXDNZE8KRZ9W9AKZAXFEQXPG",
	"type": "flow.subflow",
	"name": "Greeting Agent",
	"flowId": "01KXDPRR0NT39S75V942MEV34X",
	"position": { "x": 625, "y": 44 }
}
```

The compiler resolves that occurrence's named ports from the referenced definition's `inputs` and `outputs`. No app-specific wrapper block is required. The child graph connects its `flow.input` boundary through its implementation to its `flow.output` boundary, so changing the child's public contract updates how every parent renders and validates that subflow.

At runtime the parent step enters `waiting`, a linked child run executes through the queue, and the child output or failure resumes the parent. Execution snapshots include the resolved occurrence contracts and every nested definition. Original-definition replay therefore follows the historical child graph, while latest-definition replay resolves the current child source.

## Running And Replaying

```ts
const run = await app().flows.run(flowId, {
	name: 'Ada',
});

const details = await app().flows.runDetails(run.id!);

await app().flows.replay(run.id!, {
	definition: 'original',
});
```

The runtime captures JSON input and output at every boundary with a default one-megabyte limit. This first version does not redact values automatically, so applications must not pass credentials or unfiltered secrets through captured flow values.

## Definition Providers

`FileFlowDefinitionStore` recursively reads `flow.json` and `*.flow.json`, writes tab-formatted deterministic JSON atomically, and uses SHA-256 revisions for optimistic saves. A future database provider should implement the same `FlowDefinitionStore` contract and store the same definition shape; the runtime must continue to execute definitions rather than a separate UI representation.
````

## Guidance for AI tools
Use the documented public import `@db3.ai/app/flows` and its exported types. Prefer the source-backed examples and behavioural outcomes above over invented APIs or source-relative internal imports.