## Data directory

All state lives in `connector.data_dir` (default: the directory containing the config
file; env `OPEX_DATA_DIR`; `/data` in Docker; `/var/lib/opex-flow` under systemd):

| File | Purpose |
| --- | --- |
| `data.db` | Local history (SQLite, WAL) |
| `push_queue.db` | Durable push queue + `push_dead` dead letters |
| `dedupe.db` | Keys seen by [dedupe](/docs/flows#dedupe-idempotency) nodes (TTL 7 days by default); survives restarts |
| `reconcile.db` | Hub-side ledger entries of [ledger](/docs/flows#reconciliation-ledger-and-reconcile) nodes, reconciliation reports and the exception cases (comments and history) worked in the [workbench](/docs/finance-controls#exceptions-workbench); back it up like the other databases |
| `approvals.db` | Records parked by [approve](/docs/finance-controls#approve-node-and-approvals-inbox) nodes (durable, survives a restart) and go-live sign-off requests with their decisions |
| `pipeline_traces.db` | Pipeline execution traces, the pipeline [dead-letter queue](/docs/operations#dead-letters-correct-and-replay) and its audit trail |
| `api_token` | Generated API token (`0600`); an admin token |
| `auth.db` | SSO users, named API tokens (SHA-256 hashes only) and browser sessions (`0600`); see [Security](/docs/security#users-roles-and-single-sign-on) |
| `audit.db` | Append-only [audit log](/docs/security#audit-log) (who / when / action / target / result / request id), 5 years by default; back it up like the other databases |
| `license.json` | Cached license activation |
| `flow.json` | The deployed flow document (sources → nodes → sinks) and its version; written atomically on every deploy. Authoritative once present: the config's `transform` routes only seed it on first start (the old `pipelines.runtime.json` / `flows.layout.json` are then kept as `*.migrated`) |
| `sources.runtime.json`, `mapping.json`, `cursors.json` | Runtime edits made through the UI/API; override config on restart |

If persistence is requested (e.g. `push_queue.enabled: true`) and the directory is not
writable, startup **fails** with an error instead of silently running volatile.

## Retention

```yaml
storage:
  retention_days: 7     # default
  max_size_mb: 0        # 0 = no cap
```

Cleanup runs hourly in 10k-row batches. The file does not shrink on disk after deletes
(SQLite free pages are reused); to reclaim space stop the service and run
`sqlite3 data.db VACUUM`.

## Backup and restore

A WAL database cannot be copied safely with `cp data.db` (the `-wal` file holds recent
writes). Use SQLite's online backup:

```bash
sqlite3 /var/lib/opex-flow/data.db ".backup '/backups/data-$(date +%F).db'"
sqlite3 /var/lib/opex-flow/push_queue.db ".backup '/backups/push_queue-$(date +%F).db'"
```

Restore with the service stopped, then delete any stale `-wal`/`-shm` files. Back up the
config and `/etc/opex-flow/env` separately.

## Sizing

[INFERENCE: planning figures, not benchmarks] A canonical record is roughly 0.5-1 KB in
SQLite. 50 machines polled every 5 s is ~10 rows/s, ~0.9 M rows/day, i.e. ~0.5-1 GB/day
before the 7-day retention: set `retention_days` or `max_size_mb` to fit the disk.
The push queue is bounded by `max_quota_mb` (default 500): size it as
`records/s x 1 KB x outage seconds`. Each scripted JavaScript route runs its own QuickJS
worker limited by `transform.memory_limit_mb` (default 8 MB): budget workers x limit.
Use the deadband / aggregation `preprocess` options to cut push volume.

## Troubleshooting

| Symptom | Cause / fix |
| --- | --- |
| `401 unauthorized` from the API/UI | Missing/old token; read `<data_dir>/api_token`. |
| UI unreachable from another host | Default bind is `127.0.0.1`; set `bind_addr` / use a proxy. |
| Pushes 404 and buffer | `push_endpoint` needs the `/api/inbound-webhooks/receive/<token>` path. |
| Startup fails: data_dir not writable | Fix ownership / `StateDirectory`; durable features refuse to start volatile. |
| Config error `${VAR}` not set | Put the variable in `/etc/opex-flow/env` or the container environment. |
| `/metrics` empty, no Observability data | Binary built without the `telemetry` feature. |
| Restart loop on license error | Check `license.server_url` (`https://flow.opexcg.com`) and key; the unit stops retrying after 5 starts / 5 min. |
| FOCAS adapter offline | In `mode: library` the Fanuc library did not load (static musl, aarch64 and macOS builds cannot load it): use `mode: native` or `auto`, or the `linux-x86_64-gnu` (glibc) release. The load error is in the adapter's `last_error` and the log. |
| Build fails with `--all-features` | `focas-bundled` needs `OPEX_FOCAS_DLL_PATH`; build default features plus `telemetry,pi`. |
| Records dropped by a script | Check `transform.on_error`, the traces and the [dead-letter queue](#dead-letters-correct-and-replay) (`GET /api/pipeline/dead-letters`). |
| Records silently missing | A record no source node of the flow matches is dropped and counted `unrouted` (live stats on `WS /api/ws/flow`, `engine.stats.unrouted` in `GET /api/status`): add a source node with a matching `match` (or `*`). |
| Pushes stuck in `push_dead` | Receiver returned a permanent 4xx; fix, then `POST /api/push/dead/replay`. |

## Pipeline executor settings

Under `transform.executor`:

| Key | Default | Meaning |
| --- | --- | --- |
| `transform.executor.sink_timeout_ms` | 5000 | Per-sink deadline; sinks fan out concurrently. |
| `transform.executor.http_host_allowlist` | deny all | Hosts http nodes may call (exact host or `.suffix`); redirects are not followed. |
| `transform.executor.http_env_allowlist` | `[]` | `${ENV}` names http nodes may expand: template values, `auth` credentials and a `mode: ${NETSUITE_MODE}` switch. |
| `transform.executor.http_max_response_bytes` | 1048576 | Response cap for http nodes. |

Records a node terminates for good go to the dead-letter queue (`pipeline_dead_letters`, see
[Dead letters: correct and replay](#dead-letters-correct-and-replay)). Ordering is FIFO per route key (`{source}:{machine.id}`) with backpressure instead
of dropping. Static analysis rejects unwired branch ports, unknown sink names, edges out of
sinks, unresolved OpenAPI refs and graphs whose worst-case steps exceed `max_node_steps`
(10k). Push-error alerts count the last 10 minutes; a `pipeline` alert fires when frames were
lost or failed in that window. http node templates percent-encode frame values in URLs and
JSON-escape them in bodies (write `"${machine.id}"` quoted); non-2xx responses fail the node.

## Dead letters: correct and replay

Every record a node ends with a **permanent failure** (`Fail`: a refused NetSuite post, a script error,
an invalid frame at a sink, a dedupe key that cannot be read, a ledger node that cannot read its amount)
is parked in the dead-letter queue, unless the node's `error` port is wired
([Error branches](/docs/flows#error-branches)). Transient failures (a remote 5xx, a timeout) are **not**
parked: a source that can redeliver does so (NATS, a webhook sender's own retry). With a webhook on
`respond: "on_receipt"` nobody retries, so prefer `on_done` where a lost record matters.

Rows live in `pipeline_traces.db` (so they survive restarts) and keep:

| Field | Meaning |
| --- | --- |
| `node_id` | The node that failed (empty on rows parked before the queue tracked it). |
| `pipeline` | Match key of the source the record came through (`webhook:doku`). |
| `value` | The record as the failing node received it. |
| `origin` | The record as the source delivered it, before any node touched it. |
| `edited_value`, `edited_origin` | An operator's corrections; the originals stay. |
| `status` | `open`, `replayed`, `discarded` or `superseded` (a replay failed again; the newer row carries `replay_of`). |

Retention: open rows are kept until the table holds 10 000 rows (replayed / discarded / superseded rows are
evicted first), resolved rows for 30 days. Documents over 256 KB are replaced by a marker and cannot be
replayed until corrected.

**Operations > Dead letters** in the editor (also the `Dead letters` item in the user menu and the command
palette) lists them with filters, shows both records, edits either as JSON, and replays or discards after a
confirmation. **Replay all / Discard all** act on the current filter, oldest first, at most 500.

| Action | REST | Notes |
| --- | --- | --- |
| List | `GET /api/pipeline/dead-letters?status=&source=&node=&q=&since=&until=&limit=&offset=&include=values` | `status` defaults to `open` (`all` for everything). Returns `{total, counts, items}`; documents only with `include=values`. |
| Inspect | `GET /api/pipeline/dead-letters/{id}` | Both records, the corrections, replay attempts and the audit trail. |
| Edit | `PUT /api/pipeline/dead-letters/{id}` `{value?, origin?}` | Each a full JSON object. Open rows only (`409` otherwise). |
| Replay | `POST /api/pipeline/dead-letters/{id}/replay` `{to?, node?, wait_ms?}` | `to: "source"` (default) runs the whole flow again from the source key with the (edited) `origin`; `to: "node"` starts at `node` (default the failing one) with the (edited) `value`. Waits for the flow like `/api/ingest`: `result` is `done`, `failed`, `retry` or `pending`. |
| Discard | `POST /api/pipeline/dead-letters/{id}/discard` `{reason?}` | |
| Bulk | `POST /api/pipeline/dead-letters/replay`, `/discard` | By `ids` or by filter (`source`, `node`, `q`, `since`, `until`, `status`, `limit` up to 500); with neither, `confirm_all: true` is required. Bulk replay takes `to` and `to_node`. |
| Audit | `GET /api/pipeline/dead-letters/audit?dead_letter_id=&limit=` | Who, when, what. |

A replayed record carries `_replay {dead_letter_id, at}`. A successful replay turns the row `replayed`
(also when the flow finishes after `wait_ms`); a failing one leaves it open with the reason and the new
dead letter supersedes it, so the queue shows one row per stuck record.

Things to know before replaying:

- A replay has **real effects**: the record is posted, tallied or printed like any other.
- After a fan-out (`map`), replaying into the **source** repeats the siblings that already succeeded.
  Replay into the failing **node** instead.
- A [dedupe](/docs/flows#dedupe-idempotency) node lets a replay pass: the failed delivery never committed
  its key. A replay into the source enters the pipeline of the recorded match key; if a later deploy
  removed that source, the record goes wherever `source:machine` routes now, or is refused (`422`) when
  nothing does: replay into a node instead.
- Replay into a node needs that node to be deployed (`422` otherwise); the pipeline of the recorded
  source is preferred when several contain it.

**Audit.** Every edit, replay, discard and bulk action is written to `pipeline_dead_letter_audit` in the
same database: time, actor, interface (`api`, `mcp`), action and what: an edit lists each changed path with
before and after, a replay its target and result. The actor is the **authenticated identity** of the caller (an
SSO user's name, an API token's name, or `api-token` for the legacy token), never a label the client sends:
see [Users, roles and audit](/docs/security#users-roles-and-single-sign-on). Replay, discard and correct need the
`operator` role. The trail is kept (capped at 100 000 rows); the same actions are also in the connector-wide
[audit log](/docs/security#audit-log) (5 years by default).

## Reconciliation reports

The `reconcile` node ([Reconciliation](/docs/flows#reconciliation-ledger-and-reconcile)) compares two
record sets of a day, for example what the hub received from DOKU / GlobalTix against what was posted to
NetSuite. Its reports and the hub-side ledger live in `reconcile.db` (reports 400 days, ledger rows per
node `retention_days`, default 90).

| What | REST |
| --- | --- |
| Reports, newest first | `GET /api/reconcile/reports?name=&day=&ok=&limit=&offset=` |
| One report with its exceptions | `GET /api/reconcile/reports/{id}` |
| CSV export of the exceptions | `GET /api/reconcile/reports/{id}/csv` (`text/csv`, a download filename) |
| What the ledger holds per ledger, side and day | `GET /api/reconcile/ledger?ledger=&day=`, `GET /api/reconcile/ledger/entries?ledger=&side=&day=&limit=` |

**Operations > Reconciliation** in the editor lists the reports, shows the exceptions by kind and downloads the CSV.

## Version history

Every successful deploy (editor, API, MCP, rollback, promotion, import) is stored as an immutable **version** in
`<data_dir>/versions.db`: number (the flow `version`), time, **actor** (the authenticated identity: user, token
name or `api-token`), **message** (the optional `message` of `PUT /api/flow`, up to 500 characters), kind
(`deploy`, `rollback`, `promotion`, `import`, `baseline`), the complete flow document, the user-defined event
types and lookup-table schemas (shipped defaults and table rows are not stored), a SHA-256 content hash and, for
a promotion, where it came from. The document that was already live when history started is recorded as a
`baseline` version. Rows are never updated; `versions.keep: N` (default: keep all) prunes the oldest beyond N, never the live one.

| What | REST |
| --- | --- |
| History, newest first | `GET /api/versions?limit=&before=` |
| One version with document and definitions | `GET /api/versions/{n}` |
| What changed | `GET /api/versions/diff?from=N&to=M` (`to` defaults to the live flow) |
| Roll back | `POST /api/versions/{n}/rollback` with `{message?, if_version?}` (editor, audited `flow.rollback`) |
| Deploy with a reason | `PUT /api/flow` with `{flow, if_version, message}` |

The diff is node, edge and field level: `nodes.added/removed/changed` (a changed node lists
`changes:[{field, from, to, cosmetic}]`; `layout_only` means only x/y/size/colour moved), `edges`, `flows` (tabs),
`doc` (other top-level fields), and the changed names in `catalogue` and `lookups`, plus a `summary` of counts.

A **rollback** redeploys version N as a **new** version N+1 whose flow document equals N's; the event types and
lookup-table schemas of N are re-applied (never deleted). It fails with 422 when N no longer compiles here (for
example it uses a `${var:NAME}` that is not defined any more) and is allowed in a protected environment.
In the editor, **History** (sidebar) lists versions, shows the diff with the changed nodes highlighted on the
canvas and rolls back after a confirmation. MCP: `list_versions`, `get_version`, `diff_versions`, `rollback`;
`deploy_flow` and `deploy_draft` take a `message`.

When several people edit together ([Editing together](/docs/collaboration)), the version is created from the shared draft: the deployer is the
`actor` and everyone else who changed the draft since the previous version is listed in `co_authors` (`GET /api/versions`, History panel).
The draft lives in `<data_dir>/flow.draft.json` next to `flow.json`; back it up with the data directory.

## Environments

```yaml
environment:
  name: prod          # dev | staging | prod | any short label
  color: "#dc2626"    # optional; default by name (prod red, staging amber, dev green)
  protected: true
```

`GET /api/environment` returns `{configured, name, color, protected}`. The editor always shows a coloured badge and
stripe with the name (prod is red). With `protected: true`:

- a direct deploy (`PUT /api/flow`, MCP `deploy_flow` / `deploy_draft`, `opex-flow import`) is refused with
  **403** `{code: "protected_environment", environment}`;
- the editor is **read-only** with a banner ("This is the PROD environment: direct deploys are disabled. Promote
  from another environment or roll back.");
- **rollback** and an **approved promotion** still deploy, so production only changes through reviewed paths.
- a `${var:}` posting mode (an `http` node's `mode`) cannot be switched to `live` directly either: it needs a go-live
  request by an editor and the approval of a second, designated person (**409** `go_live_required`; see
  [Finance controls](/docs/finance-controls#go-live-gate-for-posting)).

## Variables

`${var:NAME}` is replaced by the value this instance defines, so the **same flow document runs in every
environment** and only the instance differs. Values come from the config `vars:` map (read-only through the API)
and `<data_dir>/vars.json` (`{"vars": {"NAME": "value"}}`, edited by an admin in **Variables** or with
`PUT /api/vars/{name}` `{value}`; a hand edit is noticed within a second; a `vars.json` entry shadows a config
entry). Names are `[A-Za-z_][A-Za-z0-9_.-]{0,63}`; a value is one line of plain text (no `${`).

```yaml
vars:
  ERP_BASE: https://erp-staging.example.com
  POST_MODE: sandbox_only
```

An `http` node takes `${var:NAME}` wherever it takes `${ENV}`, **without** an allowlist entry: `url`, `headers`,
`body`, `mode` (`"${var:POST_MODE}"`) and the non-secret `auth` fields (`token_url`, `client_id`, `username`, `scope`,
`realm`). Variables are **not secret** (listed to editors, exported by name): credentials stay `${secret:name}`,
and a secret `auth` field refuses a variable.

Deploy and `POST /api/flow/validate` report a reference to an **undefined variable** as an error on the node
(`variable ${var:X} is not defined in this environment`, 422 on deploy); `GET /api/flow` warns about it for the
live document. `GET /api/vars` (editor) lists the variables with `source` (`config`, `file`, `file_override`),
`names_referenced` and `undefined`. Setting and deleting are admin-only and audited (`var.set`, `var.delete`; the
value is never logged).

## Promotion

A promotion carries a version from one environment to another, signed, and waits for approval on the target.

**Source** (e.g. dev): create the signing key, list the target.

```bash
opex-flow promotion keygen --config config.yaml      # writes <data_dir>/promotion.key (0600), prints the public key
opex-flow promotion pubkey --config config.yaml
```

```yaml
promotion:
  name: dev-line1                 # default: connector.name
  targets:
- { name: staging, url: "https://flow-staging.example.com", token: "${STAGING_TOKEN}", environment: staging }
```

`token` is a named API token of the target with at least the `editor` role. In the editor, **Promote to staging**
(TopBar menu, or on a version in History) calls `POST /api/promotions/send` with `{target, version?, message?,
include_lookup_data?}`. The bundle holds the flow document, the user-defined event types, lookup-table schemas
(rows only with `include_lookup_data`), the source name, environment, version and content hash, signed with
**Ed25519** over its canonical JSON.

**Target** (e.g. staging): trust the source and choose who approves.

```yaml
promotion:
  approver_role: admin            # or editor
  trusted_sources:
- name: dev-line1
  public_key: "<base64 printed by keygen>"
```

`POST /api/promotions` verifies the bundle: the source must be a trusted source, the signature must match the
**configured** key, the hash must match the content (`401 bad_signature`, `401 untrusted_source`, `400 bad_bundle`;
nothing is stored). A good one lands in the **inbox** as `pending`; the live flow does not change.

| What | REST |
| --- | --- |
| Inbox | `GET /api/promotions?status=pending\|approved\|rejected&limit=` |
| One, with the diff against the live flow, `missing_vars`, `can_decide` | `GET /api/promotions/{id}` |
| Approve (role ≥ `approver_role`) | `POST /api/promotions/{id}/approve` `{message?}` |
| Reject (reason required) | `POST /api/promotions/{id}/reject` `{reason}` |
| Configuration (never tokens) | `GET /api/promotion` |

**Approve** applies the definitions and deploys the document as a new version of kind `promotion` with provenance
(source environment, version, promotion id), also in a protected environment. A variable the document uses that
the target does not define blocks approval with **422** `{code: "missing_vars", missing_vars: [...]}` and the
promotion stays pending. Send, receive, approve and reject are audited (`promotion.send|receive|approve|reject`).

## Git export, import and CI tests

```bash
opex-flow export --dir ../flows-repo --config config.yaml          # offline, reads the data dir
opex-flow import --dir ../flows-repo --config config.yaml --dry-run # validate only
opex-flow import --dir ../flows-repo --url https://flow-dev:3000 --token "$T" --message "from git"
opex-flow test   --dir ../flows-repo --junit report.xml            # exit 0 pass, 1 fail, 2 usage/IO
```

Layout (stable, diff-friendly: pretty JSON, sorted keys, LF):

```text
flows/<flow-id>.json         one file per tab: {flow, nodes[sorted by id], edges[sorted]}
document.json                {flow_order, extra}
catalogue/<type>.json        user-defined / overridden business-event types
lookups/<table>.schema.json  lookup-table schemas, no rows
vars.example.json            {"NAME": ""}: variable names, never values
apps.json                    when the instance serves apps
tests/*.json                 test cases (never touched by export)
```

`export --version N` writes a recorded version instead of the live flow. `import` validates first (every source
compiles, undefined variables, event types, lookup schemas) and exits 1 on any error. With `--url` it uses the REST
endpoints (a protected environment answers 403); without it, it deploys **offline** into the data dir exactly like a
live deploy (kind `import`, actor `cli:<user>`) and is refused in a protected environment; stop the connector or
restart it afterwards, it does not watch `flow.json`. Export, import, export is byte-identical, and the imported
document is the canonical order of the original (flows in order, nodes sorted by id).

**Auto-commit:** `git: {auto_commit: true, repo: /srv/flows-repo}` writes the layout into that repository after every
deploy and commits it with the `git` CLI (message = the version message or `Deploy vN`, plus `Deployed-by: <actor>`
and `Version: N`; author = the actor). A missing `git` or a non-repository is a warning on the deploy, never a failure;
nothing is committed when nothing changed.

**Test cases** (`tests/*.json`; one object, an array or `{"cases": [...]}`), run through the same dry-run engine as
`POST /api/flow/test`: no network, no real sink.

```json
{
  "name": "tags the record",
  "source": "s1",
  "input": { "source": "modbus", "timestamp": "2026-10-30T10:00:00Z", "machine": { "id": "A" }, "status": { "state": "running" }, "axes": [], "metadata": {} },
  "expected": {
"nodes": { "t": "ok", "k": "ok" },
"output": { "k": { "metadata": { "tag": 1 } } },
"sink_effects": [ { "node_id": "k", "sink": "hub", "count": 1 } ],
"failed": false
  }
}
```

`source` is a source node id or its `match` key. `expected.nodes` maps node ids to `ok`, `drop`, `error` or `fail`
(every execution must have that status). `output` gives per node a JSON subset (what a sink would deliver, else the
node's output). `sink_effects` entries must each match an effect (`count` = exactly that many). `failed` asserts whether
any node failed. Subset semantics: every expected key must match recursively, extra actual keys are fine, an expected array
matches the first elements by index, and `"$exact": true` in an object makes it exact. Optional `lookups: {table: [{key,
values}]}` seeds the offline lookup tables for the case. Variables come from the config `vars:`, `--vars file` and
`tests/vars.json`. `--junit` writes JUnit XML for CI.