This page is the contract for integrators receiving data from `connector.push_endpoint`.

## Record schema (`CanonicalMachineData`, telemetry.v1)

```json
{
  "source": "modbus",
  "timestamp": "2026-08-01T09:30:00Z",
  "machine": { "id": "PLC-01", "name": "Assembly Line PLC" },
  "status": { "state": "running", "alarmCode": "E102", "alarmMessage": "spindle overload" },
  "axes": [ { "name": "X", "position": 12.5, "unit": "mm" } ],
  "spindle": { "speed": 8000, "load": 41.5 },
  "quality": "good",
  "valueQuality": { "spindle.load": "uncertain" },
  "metadata": { "temp": 61.2 }
}
```

| Field | Type | Notes |
| --- | --- | --- |
| `source` | string, required | Origin adapter: `modbus`, `opcua`, `http`, `mqtt`, `focas`, `inject`, … |
| `timestamp` | RFC 3339 string, required | Read time at the source. |
| `machine.id` | non-empty string, required | Stable asset id. `machine.name` optional. |
| `status.state` | `running` \| `idle` \| `alarm` \| `offline` \| `unknown`, required | `alarmCode`, `alarmMessage` optional strings (camelCase on the wire). `offline` = source unreachable; `unknown` = source answered but the state is not known (MTConnect `UNAVAILABLE`, an unreadable OPC-UA/Modbus/EtherNet/IP state tag). |
| `axes[]` | optional | `{name, position, unit}` all required. |
| `spindle` | optional | `speed` required; `load` optional. |
| `quality` | `good` \| `uncertain` \| `bad`, optional | Overall data quality of the record. Absent = not stated (treat as `good`). |
| `valueQuality` | optional object | Per-value overrides of `quality`, keyed by the telemetry.v2 parameter name: `status.state`, `status.alarmCode`, `status.alarmMessage`, `axis.<name>.position`, `spindle.speed`, `spindle.load`. Each value is `good` \| `uncertain` \| `bad`. |
| `metadata` | optional object | Free-form: any JSON. Backfilled records carry `metadata.backfill = true`. |

The schema sets `additionalProperties: false` everywhere except `metadata`: a script that
adds a new top-level field produces an invalid record (handled per `transform.on_error`).
Put custom values under `metadata`. The canonical schema is published in the
`opex-contracts` repo (`schemas/telemetry.v1.json`, since contracts `v0.4.0`).

`quality` and `valueQuality` are additive and optional, but because of `additionalProperties: false`
a receiver still validating against contracts `v0.3.0` rejects records that carry them (or
`status.state: "unknown"`): upgrade receivers to `v0.4.0` before upgrading the connector.

### Data quality

| Source | Record `quality` |
| --- | --- |
| OPC-UA | Per node from the status code: `Good`, `Uncertain` for uncertain-status nodes; bad-status nodes are omitted and listed in `metadata.opcua_bad_nodes`. Record = worst carried value; a state-mapped node that yields no state gives `status.state: "unknown"` with `valueQuality["status.state"] = "bad"`. |
| PI Web API | From the event flags: `Good`; `Questionable`/`Substituted` → `uncertain`; `Good: false` or a failed point → `bad`. Per-point values stay in `metadata.parameters[].quality`; the record is `good` when all are good, `bad` when none is usable, else `uncertain`. |
| MTConnect | `UNAVAILABLE` availability/execution → state `unknown`, quality `bad`. No explicit execution state (Running assumed) → `valueQuality["status.state"] = "uncertain"`. Otherwise `good`. |
| Modbus, EtherNet/IP | Decoded values `good`; an unreadable state-mapped register/tag → state `unknown`, `valueQuality["status.state"] = "bad"`. |
| FOCAS | `good` (status read succeeded; failed optional reads are listed in `metadata.read_warnings`). |
| MQTT, NATS (wrapped values) | `good`; `bad` when the payload does not parse as its `data_type` (value `null`/raw string, `metadata.error` set). |
| HTTP, ClickHouse, PostgreSQL, inject | `good` for a fetched/injected record. |
| Any source, failed read | The offline record has `quality: "bad"`. |

NATS canonical records pass through with whatever quality the producer set. The push to
`batch.format: opexmx` maps `quality`/`valueQuality` onto the per-reading `quality` string.

## Request formats

**Single record** (default): body is the record JSON, headers `Content-Type: application/json`
and the signature headers above.

**Batch, `batch.format: hmac`:** body is
`{"connector":"<name>","sequence":N,"records":[ ...records ]}` with the signature headers and
`X-Batch-Id: <connector>:<N>`. Retries are byte-identical.

**Batch, `batch.format: opexmx`:** `Authorization: Bearer <batch.api_token>`, body
`{"sequence":N,"backfill":false,"metadata":{"sentAt":"…"},"readings":[…]}`, `X-Batch-Id`,
no body signature. `sequence` is the dedupe key.

Batching requires `push_queue.enabled: true`.

## Verifying the signature

Every push carries three signature headers (batches included, except `format: opexmx`):

| Header | Value |
| --- | --- |
| `X-Signature` | `sha256=` + hex HMAC-SHA256 of the **raw body** (legacy, no freshness). |
| `X-Opex-Timestamp` | Unix seconds at **send time** (re-signed on every retry). |
| `X-Opex-Signature` | `sha256=` + hex HMAC-SHA256 of `"{timestamp}.{body}"`. |
| `X-Opex-Key-Id` | Optional (`connector.hmac_key_id`), lets you overlap two secrets during rotation. |

Prefer the timestamped signature and reject requests outside a replay window (±5 minutes).
Always sign the exact raw bytes received, never re-serialized JSON, compare in constant
time and check lengths first.

```javascript
// Node.js (Express: use express.raw({ type: "application/json" }))
const crypto = require("crypto");
function verify(rawBody /* Buffer */, headers, secret, windowSecs = 300) {
  const ts = Number(headers["x-opex-timestamp"]);
  const sig = headers["x-opex-signature"];
  if (!Number.isFinite(ts) || typeof sig !== "string") return false;
  if (Math.abs(Date.now() / 1000 - ts) > windowSecs) return false;
  const expected = Buffer.from("sha256=" + crypto.createHmac("sha256", secret)
.update(Buffer.concat([Buffer.from(ts + "."), rawBody])).digest("hex"));
  const got = Buffer.from(sig);
  return got.length === expected.length && crypto.timingSafeEqual(got, expected);
}
```

```python
# Python (Flask: request.get_data())
import hmac, hashlib, time

def verify(raw_body: bytes, headers, secret: str, window: int = 300) -> bool:
ts, sig = headers.get("X-Opex-Timestamp"), headers.get("X-Opex-Signature")
if not ts or not sig or not ts.isdigit() or abs(time.time() - int(ts)) > window:
    return False
mac = hmac.new(secret.encode(), ts.encode() + b"." + raw_body, hashlib.sha256)
return hmac.compare_digest(sig, "sha256=" + mac.hexdigest())
```

```go
// Go
func verify(raw []byte, ts, sig, secret string, window time.Duration) bool {
	n, err := strconv.ParseInt(ts, 10, 64)
	if err != nil || time.Since(time.Unix(n, 0)).Abs() > window {
		return false
	}
	mac := hmac.New(sha256.New, []byte(secret))
	mac.Write([]byte(ts + "."))
	mac.Write(raw)
	return hmac.Equal([]byte(sig), []byte("sha256="+hex.EncodeToString(mac.Sum(nil))))
}
```

## Delivery semantics

- **At-least-once, ordered.** A timed-out request may have been processed; the retry
  produces a duplicate. Single-record pushes have **no idempotency key**: dedupe on
  `(machine.id, timestamp, source)`. Batches carry `X-Batch-Id` / `sequence`.
- **Duplicates also originate upstream of the push.** NATS JetStream ingest is acknowledged
  only after the record is handled downstream (for the push sink: stored in the push queue),
  so a failed sink, a restart or a slow pipeline redelivers the source message and the same
  record can be pushed twice. A `map` fan-out whose children partly failed re-runs the
  children that had succeeded. Dedupe on `(machine.id, timestamp, source)`; see "NATS
  delivery semantics" in the sources guide.
- **Retry:** `408`, `429`, `5xx` and transport errors back off exponentially (honoring
  `Retry-After`, capped at 300 s). `401`/`403` are treated as credential problems: the
  connector backs off and keeps the row. `409` means "already have it" and is acknowledged.
  Any other `4xx` is permanent: the row moves to the dead-letter table (`GET /api/push/dead`,
  replay with `POST /api/push/dead/replay`, purge with `DELETE /api/push/dead`; queue
  health at `GET /api/push/status`).
- **Durability:** without `push_queue.enabled` the retry buffer is in memory (1000
  entries, oldest evicted) and lost on restart. Enable the durable queue for
  store-and-forward across restarts and outages; it is bounded by `max_quota_mb`
  (default 500) and evictions are counted.
- **Preprocessing** (deadband / aggregation) only affects the push path.

## Replay protection

`X-Opex-Signature` binds a send-time timestamp to the body, so a captured request is only
replayable inside your window. Enforce the window, TLS, and still dedupe on
`(machine.id, timestamp, source)` / `X-Batch-Id`, since an unacknowledged request is
legitimately re-sent (with a fresh timestamp). `batch.format: opexmx` is authenticated by
the bearer API key instead of a body signature. To rotate `hmac_secret`, set
`connector.hmac_key_id`, accept both keys on the receiver keyed by `X-Opex-Key-Id`, then
retire the old one.