# Flue persistence reference

> `PersistenceAdapter` contract, store interfaces, built-in database adapters, and `db.ts` discovery for durable execution.

- Repository: withastro/flue-with-vercel-eve
- GitHub: https://github.com/withastro/flue
- Human docs: https://grok-wiki.com/public/docs/withastro-flue-with-vercel-eve-f4b79875fff6
- Complete Markdown: https://grok-wiki.com/public/docs/withastro-flue-with-vercel-eve-f4b79875fff6/llms-full.txt

## Source Files

- `withastro-flue:packages/runtime/src/adapter.ts`
- `withastro-flue:packages/postgres/src/index.ts`
- `withastro-flue:packages/mysql/README.md`
- `withastro-flue:packages/libsql/README.md`
- `withastro-flue:blueprints/database.md`
- `withastro-flue:CHANGELOG.md`

---

---
title: "Flue persistence reference"
description: "`PersistenceAdapter` contract, store interfaces, built-in database adapters, and `db.ts` discovery for durable execution."
---

Flue persists agent sessions, submission lifecycles, workflow-run records, and event streams through a single `PersistenceAdapter` contract. On the Node target, you configure persistence by default-exporting an adapter from a source-root `db.ts` file; the build discovers it, wires it into the generated server, and calls `migrate()` then `connect()` once at startup. Without `db.ts`, Node falls back to in-memory SQLite that lasts only for the process lifetime.

## What persistence stores

A Flue database holds **runtime state**, not application business data.

| Stored by Flue | Not stored by Flue |
| --- | --- |
| Agent session messages, compaction state, and persisted image chunks | Sandbox files and installed dependencies |
| Accepted direct prompts and `dispatch(...)` submissions, including turn journals, claims, and leases | External API side effects |
| Workflow-run records and persisted run events | Customer records, tickets, payments, or other app-owned data |
| Run indexing for `/runs` lookups and `listRuns()` | Provider credentials or secrets |

Durable persistence enables interruption recovery: accepted work survives process restart, leases coordinate concurrent coordinators, and turn journals back replay-safe recovery. See [Runtime models](/runtime-models) for how sessions, runs, and dispatch receipts relate to durable execution.

## `db.ts` discovery and wiring

`db.ts` is a **Node-target** concern. Cloudflare uses Durable Object SQLite automatically and **rejects** `db.ts` at build time.

### File location

Place exactly one `db.ts` in the project source root. Flue resolves the source root in this order:

1. `<project-root>/.flue/`
2. `<project-root>/src/`
3. `<project-root>/`

Extension priority matches other optional entries: `db.ts` > `db.mts` > `db.js` > `db.mjs`.

The file must **default-export** a `PersistenceAdapter`. Blueprint-generated files include a first-line marker such as `// flue-blueprint: database/postgres@1`.

### Build-time and startup lifecycle

During `flue build`, the Node plugin discovers `db.ts` and imports it into the generated server entry. At server startup:

<Steps>
<Step title="Validate the export">

The generated entry checks that the default export exposes `connect()`.

</Step>
<Step title="Run migrate()">

When present, `migrate()` runs once to create schema idempotently and stamp the schema version.

</Step>
<Step title="Await connect()">

`connect()` returns `{ executionStore, runStore, eventStreamStore }`. An unreachable or misconfigured database fails at boot, not on the first request.

</Step>
<Step title="Release on shutdown">

When present, `close()` runs during graceful shutdown to release pools and file handles.

</Step>
</Steps>

Without `db.ts`, the generated server uses `sqlite()` from `@flue/runtime/node` with an in-memory database — equivalent to omitting a file path.

Verify discovery with `flue build`: the build summary lists the resolved `database` path when `db.ts` is found.

## `PersistenceAdapter` contract

Import types and helpers from `@flue/runtime/adapter`. There is one contract for every backend — no SQL-only or expert tiers.

```ts
interface PersistenceAdapter {
  connect(): PersistenceStores | Promise<PersistenceStores>;
  migrate?(): void | Promise<void>;
  close?(): void | Promise<void>;
}

interface PersistenceStores {
  readonly executionStore: AgentExecutionStore;
  readonly runStore: RunStore;
  readonly eventStreamStore: EventStreamStore;
}
```

<ResponseField name="connect()" type="PersistenceStores | Promise<PersistenceStores>">

Opens the database and returns all three stores. Async pool setup, remote handshakes, and schema-version checks (for adapters without `migrate()`) belong here. Called once at startup.

</ResponseField>

<ResponseField name="migrate()" type="void | Promise<void>">

Optional. Brings the store to the current schema/format version before `connect()`. Creates missing schema idempotently and stamps the version on first creation.

</ResponseField>

<ResponseField name="close()" type="void | Promise<void>">

Optional. Releases connections, pools, or file handles during shutdown.

</ResponseField>

Factory functions in adapter packages (`sqlite()`, `postgres()`, `mysql()`, `libsql()`, and others) return this interface.

## Store interfaces

`PersistenceStores` bundles three stores. Each documents per-method invariants in terms of observable behavior — atomicity, idempotency, and gating — so non-SQL backends are first-class implementations.

### `AgentExecutionStore`

Groups session storage and submission lifecycle storage.

```ts
interface AgentExecutionStore {
  readonly sessions: SessionStore;
  readonly submissions: AgentSubmissionStore;
}
```

**`SessionStore`** persists complete session records:

| Method | Contract |
| --- | --- |
| `save(id, data)` | Persist the full `SessionData` under the Flue storage key. |
| `load(id)` | Return saved data, or `null` when none exists. |
| `delete(id)` | Remove the stored session record. |

**`AgentSubmissionStore`** owns ordered admission, claims, turn journals, stream chunks, attempt markers, lease renewal, and session deletion coordination for direct prompts and `dispatch(...)` input.

Stability note: `SessionStore`, `RunStore`, and `EventStreamStore` are stable. The `AgentSubmissionStore` turn-journal, stream-chunk, and lease method groups mirror the durable-execution engine and may change before 1.0.

Key invariant groups:

| Group | Behavior |
| --- | --- |
| Admission | `admitDispatch()` is idempotent by dispatch id; same id with different payload returns conflict; `admitDirect()` admits queued submissions with replay idempotency. Both throw while the session is being deleted. |
| Claims | `claimSubmission()` atomically moves queued → running only when the submission is the runnable head of its session. Concurrent claims for the same submission must never both succeed. |
| Turn journal | At most one journal per submission. Phases advance only for uncommitted journals owned by the calling attempt. `replaceTurnJournalAttempt()` is the recovery handoff. |
| Stream chunks | `appendStreamChunkSegment()` is insert-only; duplicate keys return `false` without overwriting. |
| Leases | `renewLeases()` extends expiry for running submissions owned by the coordinator. `listExpiredSubmissions()` returns running submissions past `leaseExpiresAt`. |
| Deletion | `deleteSession()` coordinates three-phase settled-state removal with a durable deletion marker. `listPendingSessionDeletions()` surfaces markers left by crashed processes. |

Default durability constants exported from `@flue/runtime/adapter`:

| Constant | Value |
| --- | --- |
| `DURABILITY_DEFAULT_MAX_ATTEMPTS` | 10 |
| `DURABILITY_DEFAULT_TIMEOUT_MS` | 3,600,000 (1 hour) |
| `LEASE_DURATION_MS` | 30,000 (30 seconds) |

### `RunStore`

Persists workflow-run records and serves lookup and listing for `/runs`, `flue logs`, and inspection primitives. Event payloads live in `EventStreamStore`.

| Method | Contract |
| --- | --- |
| `createRun()` | Persist a new `active` run. Idempotent, first-writer-wins on duplicate `runId`. |
| `endRun()` | Finalize with terminal status, result, or error. No-op when unknown. |
| `getRun()` | Full run record, or `null`. |
| `lookupRun()` | `RunPointer` projection (excludes `payload`, `result`, `error`). |
| `listRuns()` | Newest first, filterable by status/workflow, paginated via opaque cursor. |

Helpers: `encodeRunCursor()`, `decodeRunCursor()`, `DEFAULT_LIST_LIMIT`, `MAX_LIST_LIMIT`.

### `EventStreamStore`

Append-only event streams for agent instances and workflow runs. Paths are typically `agents/<name>/<id>` or `runs/<runId>`.

| Method | Contract |
| --- | --- |
| `createStream()` | Idempotent stream creation. |
| `appendEvent()` | Append and return a new Durable Streams offset. Throws on missing stream. |
| `readEvents()` | Read strictly after `offset`. Missing stream returns empty, up-to-date result (no throw). |
| `closeStream()` | Close the stream. |
| `getStreamMeta()` | Stream metadata, or `null`. |
| `subscribe()` | In-process listener for appends or closure on this store instance. |

Offsets use Durable Streams format (`<readSeq>_<seq>`) plus sentinel `"-1"`. Use `formatOffset()` and `parseOffset()` from `@flue/runtime/adapter`.

## Built-in adapters

<Tabs>
<Tab title="Node built-in">

**`sqlite(path?)`** from `@flue/runtime/node` is the zero-dependency built-in adapter. Omit the path or pass `':memory:'` for process-lifetime storage; pass a file path for restart-surviving state on a single host.

```ts title="src/db.ts"
import { sqlite } from '@flue/runtime/node';

export default sqlite('./data/flue.db');
```

Uses `node:sqlite` with WAL mode for file-backed databases. Shares the SQL implementation with Cloudflare Durable Object SQLite.

</Tab>
<Tab title="SQL packages (BYO driver)">

SQL adapter packages accept a **runner** with three functions — `query`, `transaction`, and `close` — wrapped around your configured driver. Flue does not bundle production drivers.

| Package | Placeholders | Install |
| --- | --- | --- |
| `@flue/postgres` | `$1`, `$2`, … | `flue add database postgres` |
| `@flue/mysql` | `?` | `flue add database mysql` |
| `@flue/libsql` | `?` | `flue add database libsql` or `flue add database turso` |

<CodeGroup>
```ts title="Postgres (porsager postgres)"
import { postgres, type PostgresQuery } from '@flue/postgres';
import sql from 'postgres';

const db = sql(process.env.DATABASE_URL!);

export default postgres({
  query: (text, params) => db.unsafe(text, params),
  transaction: <T>(fn: (tx: { query: PostgresQuery }) => Promise<T>) =>
    db.begin((tx) => fn({ query: (text, params) => tx.unsafe(text, params) })) as Promise<T>,
  close: () => db.end(),
});
```

```ts title="MySQL (mysql2)"
import { mysql, type MysqlQuery } from '@flue/mysql';
import mysql2 from 'mysql2/promise';

const pool = mysql2.createPool(process.env.MYSQL_URL!);

const toRows = (result: unknown): Record<string, unknown>[] =>
  Array.isArray(result) ? result.map((row) => ({ ...row })) : [];

export default mysql({
  query: async (text, params = []) => {
    const [result] = await pool.execute(text, params);
    return toRows(result);
  },
  transaction: async <T>(fn: (tx: { query: MysqlQuery }) => Promise<T>) => {
    const connection = await pool.getConnection();
    try {
      await connection.beginTransaction();
      const result = await fn({
        query: async (text, params = []) => {
          const [rows] = await connection.execute(text, params);
          return toRows(rows);
        },
      });
      await connection.commit();
      return result;
    } catch (error) {
      await connection.rollback();
      throw error;
    } finally {
      connection.release();
    }
  },
  close: () => pool.end(),
});
```

```ts title="libSQL / Turso"
import { libsql } from '@flue/libsql';
import { createClient, type ResultSet } from '@libsql/client';

const client = createClient({
  url: process.env.LIBSQL_URL!,
  authToken: process.env.LIBSQL_AUTH_TOKEN,
});

const toRows = (rs: ResultSet) =>
  rs.rows.map((row) => Object.fromEntries(rs.columns.map((col) => [col, row[col]])));

export default libsql({
  query: (text, params = []) => client.execute({ sql: text, args: params }).then(toRows),
  transaction: (fn) => client.transaction('write').then(async (tx) => {
    try {
      const result = await fn({
        query: (text, params = []) => tx.execute({ sql: text, args: params }).then(toRows),
      });
      await tx.commit();
      return result;
    } catch (error) {
      await tx.rollback();
      throw error;
    } finally {
      tx.close();
    }
  }),
  close: () => client.close(),
});
```
</CodeGroup>

MySQL requires MySQL 8 with InnoDB. For local `file:` libSQL databases, serialize operations in the runner to avoid `SQLITE_BUSY` contention.

</Tab>
<Tab title="Non-SQL packages">

Driver-free adapters for backends that are not SQL-shaped:

| Package | Install |
| --- | --- |
| `@flue/redis` | `flue add database redis` |
| `@flue/mongodb` | `flue add database mongodb` |

Blueprints also exist for Supabase (Postgres), Valkey, and custom backends via `flue add database <url>`.

</Tab>
<Tab title="Cloudflare">

No `db.ts`. Generated agent and workflow Durable Objects use SQLite automatically. Run indexing lives in the generated `FlueRegistry` Durable Object. A present `db.ts` causes the Cloudflare build to fail with an explicit error.

</Tab>
</Tabs>

### Choosing an adapter

| Use case | Adapter |
| --- | --- |
| Local dev, restart persistence optional | No `db.ts` (in-memory) or file-backed `sqlite()` |
| Single-host Node deployment | File-backed `sqlite()` |
| Multi-replica Node or host replacement | `@flue/postgres`, `@flue/mysql`, `@flue/libsql`, or hosted Turso |
| Existing operational environment | Match your ops stack (`mysql`, `redis`, `mongodb`, etc.) |
| Cloudflare deployment | Built-in Durable Object SQLite |

## Schema versioning

Every adapter must durably record its schema/format version at store creation and **fail loudly** before reading or writing when opened against an unknown or newer version.

SQL adapters use a one-row `flue_meta` table with key `schema_version`. Non-SQL adapters implement the same obligation natively.

Exports from `@flue/runtime/adapter`:

| Export | Purpose |
| --- | --- |
| `FLUE_SCHEMA_VERSION` | Current version to stamp at creation (currently `1`). |
| `assertSupportedFlueSchemaVersion(storedVersion)` | Throws on mismatch. |
| `PersistedSchemaVersionError` | Error type for version conflicts. |

`migrate()` creates schema idempotently — there is no separate migration CLI command. A database written by a newer Flue version refuses to start after a rollback rather than corrupting state.

## Custom adapters

When built-in adapters do not fit, implement `PersistenceAdapter` directly:

1. Import contract types from `@flue/runtime/adapter` only — not `@flue/runtime/internal`.
2. Honor durable submission invariants with your backend's real primitives (transactions, conditional writes, atomic counters).
3. Run `migrate()` at startup with idempotent schema/collection creation.
4. Return all three stores from `connect()`.
5. Validate with contract suites from `@flue/runtime/test-utils`:

```ts
import {
  defineEventStreamStoreContractTests,
  defineRunStoreContractTests,
  defineStoreContractTests,
} from '@flue/runtime/test-utils';

defineStoreContractTests('MyBackend', {
  async create() {
    const adapter = myBackend();
    await adapter.migrate?.();
    const { executionStore } = await adapter.connect();
    return executionStore;
  },
  async cleanup() { /* close connections, delete temp state */ },
});
```

Use `flue add database <url>` to scaffold from the generic database blueprint, or `flue update database <name|url>` to refresh an existing adapter while preserving customizations.

Adapter helpers for storage-key, timestamp, payload-validation, cursor, and offset semantics: `createSessionStorageKey()`, `parseAcceptedAt()`, `isSubmissionPayload()`, `clampLimit()`, and persisted image-chunk utilities.

## Environment and credentials

Read connection strings from the environment — commonly `DATABASE_URL`, `MYSQL_URL`, `LIBSQL_URL` — and load them through project conventions or `flue dev --env <file>` / `flue run --env <file>`. Never commit real credentials. Blueprint-generated `db.ts` files expect you to supply values through env files or your secret manager.

## Troubleshooting

| Symptom | Likely cause |
| --- | --- |
| Build fails on Cloudflare with `db.ts` error | Remove `db.ts` or move it outside the source root; Cloudflare uses DO SQLite. |
| Server fails at startup with persistence error | Database unreachable, wrong credentials, or `connect()` validation failed. Check env vars and network. |
| `PersistedSchemaVersionError` at startup | Database was written by a different Flue version. Align versions or provision a fresh store. |
| Accepted work lost on restart | No `db.ts`, or `sqlite()` without a file path (in-memory). Add file-backed `sqlite()` or a shared adapter. |
| `SQLITE_BUSY` with local libSQL file | Concurrent writes to an embedded file. Serialize runner operations or use hosted Turso/sqld. |
| Session admissions blocked after crash | Orphaned deletion marker. Coordinator resumes via `listPendingSessionDeletions()` at startup. |

## Related pages

<CardGroup>
<Card title="Flue project layout" href="/flue-project-layout">
Source-root resolution (`.flue/` vs `src/`), where `db.ts` lives, and discovery boundaries.
</Card>
<Card title="Runtime models" href="/runtime-models">
Sessions, workflow runs, dispatch receipts, and durable execution semantics.
</Card>
<Card title="Flue configuration reference" href="/flue-configuration-reference">
`flue.config.*` keys, source-root precedence, and env loading.
</Card>
<Card title="Flue workflows" href="/flue-workflows">
Workflow runs, event streams, and `flue logs` inspection.
</Card>
<Card title="Flue HTTP API reference" href="/flue-http-api-reference">
`/runs` routes, stream reads, and run metadata.
</Card>
<Card title="Flue deploy" href="/flue-deploy">
Node vs Cloudflare targets and persistence implications per platform.
</Card>
</CardGroup>