This document specifies the **STA Connector**, a standalone, on-premises agent that
bridges heterogeneous sensor networks to an [[OGC-STA-1|OGC SensorThings API v1.1]]
server. It defines the connector's **ingest contract** (HTTP, MQTT and Modbus TCP
sources), the declarative **transform intermediate representation (IR)** that a portal
authors and the connector executes, the **processing model** (normalize, validate,
batch, retry, quarantine), the **egress contract** to the STA server, the local
**management API** (source provisioning and run history), and the **tasking control
plane** that executes device-management tasks and decrypts device secrets. It is the
normative contract against which the STA Portal and operator tooling are built. The
companion server specification is [[STA-SERVER]].
This is a living technical specification maintained in the `sta-connector`
repository. It tracks the implemented behaviour of the service on its default branch;
the source code is the ultimate reference, and this document is updated as part of the
same change set whenever normative behaviour changes.
Feedback and change proposals are handled through GitHub issues and pull requests on
the repository. There is no external standards body governing this document.
## Terminology
Terms from [[OGC-STA-1]] (Thing, Datastream, Observation, FeatureOfInterest) and
[[OGC-STA-2]] (Task, TaskingCapability) retain their meaning from those specifications.
- connector: the service specified by this document; an on-premises agent
with a stable identity, modelled on the server as a [=Thing=] of `type` `connector`.
- Thing: a SensorThings Thing as defined in [[STA-SERVER]].
- source: a configured ingest-plus-egress pipeline inside the [=connector=]
process, identified by a unique `name`.
- raw event: the envelope an ingest adapter emits for one received
measurement, before normalisation.
- observation: the OGC Observations-and-Measurements record the connector
publishes to the server.
- transform graph: the declarative node-and-edge [=transform IR=] document a
portal authors to map a [=raw event=] to zero or more [=observation=]s.
- transform IR: the typed, data-only intermediate representation defined in
— nodes are operations, edges connect node ports.
- run: one reportable pipeline outcome (a batch publish, a batch failure, a
test-run, or a quarantine burst) recorded in the local run history.
- quarantine: the holding area for measurements that failed validation or
were rejected by the server with a non-retryable status.
- device secret: a credential the connector can decrypt with its private key;
the server holds only ciphertext sealed to the connector's public key.
## Data model {#data-model}
### Raw event {#raw-event}
An ingest adapter MUST emit, for each received measurement, a [=raw event=] with the
fields:
| Field | Type | Requirement |
| --- | --- | --- |
| `source_name` | string | MUST identify the originating [=source=]. |
| `value` | any JSON | The measured value (scalar or object). |
| `timestamp` | string, nullable | ISO-8601 instant; when absent the connector MUST default to wall-clock UTC at ingest time. |
| `raw` | object, optional | Provenance/debugging metadata, carried through to the Observation's `parameters.raw`. |
### Observation {#observation-model}
The connector MUST normalise each accepted measurement to an OGC
Observations-and-Measurements record with:
| Field | Type | Requirement |
| --- | --- | --- |
| `phenomenonTime` | string | UTC ISO-8601 instant. The connector MUST emit the `Z` form with no fractional seconds. |
| `result` | any JSON | The normalised value. |
| `resultTime` | string, optional | UTC ISO-8601 instant when present. |
| `parameters` | object | MUST include `source`; MUST include `raw` when the raw event carried it. |
| `Datastream` | object | MUST be the inline reference `{ "@iot.id": }`. |
The connector MUST NOT include a `FeatureOfInterest` in the payload; the server
auto-resolves it ([[STA-SERVER]]). Timestamps accepted on ingest MAY be in any ISO-8601
form; the connector MUST normalise them to UTC `Z`.
### Run history record {#run-record}
The connector MUST persist [=run=] records (see ) with at least:
`source_name`, `event`, `status`, `started_at`, `finished_at`, `duration_ms`,
`items_in`, `items_out`, a nullable `error_message`, and a redacted `log_excerpt`.
## Ingest Contract {#ingest}
This section is normative for the **Ingest Contract** class. A [=source=] declares a
`kind`; a conforming connector MUST implement at least one of the kinds below and MUST
reject an unknown kind at configuration time.
### HTTP ingest {#ingest-http}
For a `http_ingest` source the connector MUST expose `POST /ingest/{source_name}` (or the
source's configured `path`). The request body MUST be JSON of the shape:
```json
{ "value": , "timestamp": "", "raw": { } }
```
`value` is REQUIRED; `timestamp` and `raw` are OPTIONAL. The connector MUST respond
`202 Accepted` once the measurement is enqueued, `400 Bad Request` for a malformed body,
and a `5xx` status when the queue is saturated. Acceptance MUST NOT block on egress.
### MQTT subscribe {#ingest-mqtt}
For a `mqtt_subscribe` source the connector MUST subscribe to the configured sensor-side
broker and topic and emit one [=raw event=] per received message. The connector SHOULD
pool a single subscriber per distinct broker tuple `(host, port)`. Sensor-side broker
credentials are independent of the STA-side broker ().
### Modbus TCP {#ingest-modbus}
For a `modbus_tcp` source the connector MUST poll the configured register over
[[MODBUS]] at the configured interval and emit one [=raw event=] per poll. The source
configuration MUST carry `host`, `port`, `register`, `register_type`
(`holding` or `input`) and `poll_interval_seconds`, and MAY carry a `scale` factor
applied to the raw register value.
## Transform IR {#transform-ir}
This section is normative for the **Transform IR** class. A [=transform graph=] is
authored by the portal, stored on the server (in a device `Configuration`), mirrored
into source configuration, and executed by the connector. The IR is **data, not code**:
a conforming connector MUST NOT evaluate arbitrary code from a graph.
### Graph structure {#ir-structure}
A graph is a JSON document:
| Object | Fields |
| --- | --- |
| `TxGraph` | `version` (integer, default `1`), `nodes` (array of `TxNode`), `edges` (array of `TxEdge`). |
| `TxNode` | `id` (string), `type` (string), `config` (object, type-specific). |
| `TxEdge` | `id` (string), `from` (string), `to` (string). |
An edge endpoint is a `":"` reference. Most nodes have a single input port
`i` and a single output port `o`; the `build_obs` sink has input ports `val` and `t`.
Each input port MUST have at most one incoming edge.
### Evaluation semantics {#ir-evaluation}
The connector MUST evaluate the graph **lazily from each `build_obs` sink** back to its
inputs, and MUST support **fan-out**: one [=raw event=] MAY produce zero or more
[=observation=]s (one per `build_obs` node that yields a value). The connector MUST
detect cycles and fail the graph with an error rather than loop.
A node MAY produce a **drop** sentinel (e.g. a `filter` that rejects, an optional input
that is not connected). A drop MUST propagate downstream; a `build_obs` that receives a
drop (or `null`) on its `val` port MUST produce no Observation. A graph whose branches
all drop MUST yield an empty result, not an error.
### Node types {#ir-nodes}
A conforming connector MUST implement the following node types with the stated
semantics. Configuration keys are given in parentheses.
| Type | Category | Behaviour |
| --- | --- | --- |
| `raw_payload` | input | Emits the raw ingest value on port `o`. |
| `json_field` | decode | Extracts a `/a/b/0` pointer path (`path`); on a missing path either drops (`on_missing`) or errors. |
| `bytes_decode` | decode | Decodes a `hex` or `base64` string (`encoding`) to raw bytes. |
| `bytes_unpack` | bytes | Reads one struct-typed number (`type` ∈ int/uint 8–64, `float32/64`) at a byte `offset`, with `byte_order` (`big`/`little`) and optional `scale`. |
| `scale` | transform | `x · factor + offset`. |
| `offset` | transform | `x + offset`. |
| `formula` | transform | Evaluates `expr` over the single variable `x` using a **whitelisted AST** (arithmetic and the functions `min`, `max`, `abs`, `round`, `sqrt`, `pow` only). |
| `unit` | transform | Linear unit conversion (`from`→`to`) within a dimension (temperature, pressure). |
| `cast` | transform | Casts to `getal` (number), `tekst` (text), `ja/nee` (boolean) or `tijdstip` (time). |
| `clamp` | transform | Range check (`min`/`max`) with `on_out` of clamp-to-bound, skip, or invalid. |
| `round` | transform | Rounds to `decimals`. |
| `lookup` | transform | Maps `x` through `rows[{k,v}]` with a `fallback`. |
| `default` | transform | Substitutes `value` when the input is dropped or `null`. |
| `filter` | logic | Drops `x` unless `x threshold` holds (`op` ∈ `> ≥ < ≤ = ≠`). |
| `parse_time` | output | Produces a timestamp from `nu` (now) or a `path`, parsed per `format` (`ISO-8601`, `Unix (s)`, `Unix (ms)`). |
| `build_obs` | output (sink) | Emits an Observation from its `val` and `t` ports for the configured `datastream`; `time_source` `now` uses the evaluation clock. |
### Forward-compatibility {#ir-future}
A conforming connector MUST raise a clear, typed error for a node type it does not yet
implement (e.g. `modbus_reg`, `mqtt_topic`, `deadband`, `switch`) rather than silently
producing a wrong result. The `formula` evaluator MUST reject any construct outside the
whitelist (names other than `x`, attribute access, calls to non-whitelisted functions).
### Security {#ir-security}
The transform engine MUST run only on the customer's trusted network and MUST NOT
provide a path to arbitrary code execution. The `formula` node's whitelisted-AST
evaluator is the only expression surface and MUST remain limited to arithmetic over `x`
and the fixed function set.
## Processing Model {#processing}
This section is normative for the **Processing Model** class.
### Pipeline {#processing-pipeline}
Each accepted measurement MUST pass through: **normalise** (to the Observation shape of
) → **validate** → optional **transform** () →
**batch** → **egress** (). A measurement that fails normalisation or
validation MUST NOT be published and MUST be recorded as described in
.
### Validation {#processing-validation}
A source MAY declare a validation block (`min`, `max`, `type`). The connector MUST apply
it after normalisation and before batching. A value that fails validation MUST be
quarantined, not published.
### Batching {#processing-batching}
The connector MUST batch accepted Observations per source and MUST flush a batch when
either the item ceiling (`batch.max_items`) or the age ceiling (`batch.max_age_seconds`)
is reached, whichever comes first. Batching is for delivery efficiency and back-pressure;
it MUST NOT reorder Observations within a source ahead of their `phenomenonTime`
watermark.
### Retry {#processing-retry}
On an egress failure the connector MUST distinguish failure classes:
- a **4xx** rejection (other than `429`) MUST NOT be retried and MUST be quarantined;
- a **5xx** or network/connection failure MUST be retried with **exponential backoff**
(initial 500 ms, maximum 30 s, multiplier 2.0, with jitter), up to the
configured attempt ceiling;
- a **429** response MUST be honoured per its `Retry-After` header where present.
### Quarantine {#processing-quarantine}
The connector MUST hold measurements that fail validation or are rejected with a
non-retryable status in a [=quarantine=] area, and MUST expose categorical counts and reasons
through the management API. The connector MUST NOT expose raw measurement payloads
through any admin response; only categorical reasons. A sustained quarantine increase MAY
be recorded as a `quarantine_burst` [=run=].
## Egress Contract {#egress}
This section is normative for the **Egress Contract** class — the core of a
[=conforming connector=]. The wire shape and server behaviour are specified by
[[STA-SERVER]]; this section constrains the connector side.
### HTTP egress {#egress-http}
By default the connector MUST publish each Observation with a single
`POST /v1.1/Datastreams({id})/Observations` carrying the Observation body of
and the header `Authorization: Bearer `, where the token is
resolved from an environment variable (default `STA_TOKEN`) and MUST NOT be read from the
configuration file. The connector MUST capture the server-assigned `@iot.id` from the
`201` response for audit. Bulk `dataArray` egress MAY be supported behind a feature flag
but MUST default to single-POST, because the server does not implement the bulk endpoint
in this revision.
### MQTT egress {#egress-mqtt}
When configured for MQTT egress the connector MUST publish the same Observation body to
the OGC topic `v1.1/Datastreams({id})/Observations` over [[MQTT]] on the configured
STA-side broker. Until the server implements its MQTT extension, deployments point MQTT
egress at a bundled broker; switching to the server's broker MUST be a configuration
change only, with no connector code change.
### Delivery guarantees {#egress-guarantees}
The connector provides at-least-once delivery; the server defines no idempotency key, so
the connector SHOULD dedupe at the source where exactly-once matters. Timestamps MUST be
emitted as UTC `Z` with no fractional seconds for predictability.
## Management API {#management}
This section is normative for the **Management API** class. All routes mount under the
`/admin` prefix. Bodies are JSON; times are UTC ISO-8601 with a `Z` suffix. The portal
never calls these routes directly from a browser; it calls through a server-side proxy
([[STA-SERVER]]).
### Source provisioning (CONN-1) {#conn1}
A conforming Management API MUST expose source lifecycle management so a client can
provision sources without editing YAML or restarting:
| Method & path | Purpose |
| --- | --- |
| `POST /admin/sources` | Create a source from a JSON spec (`201`; `409` on name conflict; `422` on invalid config). |
| `PUT /admin/sources/{name}` | Replace a source's config (in-flight queue preserved). |
| `PATCH /admin/sources/{name}` | Partial update. |
| `DELETE /admin/sources/{name}` | Stop, drain (best-effort ≤10 s) and remove (`204`). |
| `POST /admin/sources/{name}:start` | Start a stopped source (idempotent). |
| `POST /admin/sources/{name}:stop` | Stop a running source; in-flight batch finishes. |
| `POST /admin/sources/{name}:test-run` | Execute one cycle **without** updating watermarks; returns the normalised Observation and validation outcome. |
| `POST /admin/sources/{name}:resync` | Drop the watermark and re-read from the cursor. |
| `GET /admin/sources/{name}/logs?tail=N` | Tail the in-memory per-source ring buffer. |
The colon `:action` custom-method style follows [[AIP-136]]. Every mutation MUST be
written back to `connector.yaml` **atomically** (write-temp, fsync, rename) so a restart
resumes identical state, and MUST be serialised per source. A `test-run` MUST NOT update
watermarks. Secrets MUST NOT appear in any response body.
### Source state machine {#conn1-state}
A source MUST expose one of the states `starting`, `running`, `stopping`, `stopped`,
`error`. `error` is sticky: a source in `error` MUST require an explicit `:start` after
the operator resolves the fault. A missing required environment variable MUST place the
source in `error` with a `missing_env` reason.
### Run history (CONN-2) {#conn2}
A conforming Management API MUST record reportable pipeline outcomes and expose them
read-only:
| Method & path | Purpose |
| --- | --- |
| `GET /admin/runs` | List runs across sources; filterable by `since`, `until`, `status`, `event`, `source`; paginated by `cursor`. |
| `GET /admin/sources/{name}/runs` | Runs for one source. |
| `GET /admin/runs/{id}` | A single run including the full `log_excerpt`. |
| `GET /admin/runs/stats` | Aggregated counts grouped by `day`/`hour`/`source`/`status`/`event`. |
The four reportable events are `batch_publish`, `batch_failure`, `test_run` and
`quarantine_burst`. Records MUST be persisted durably (so they survive a restart) and
MUST be written synchronously at the end of each event for post-crash forensic value.
The `log_excerpt` MUST be passed through secret redaction before storage and MUST be
capped (8 KB, truncated from the head). Records MUST be subject to a configurable
retention sweep (default 30 days).
### Health and operational views {#conn-health}
`GET /admin/health` MUST report overall status plus, per source, the lifecycle state,
the freshness watermark (`latest_phenomenon_time`), throughput and last error, and MUST
report STA-server reachability derived from publish outcomes (not an active probe) and
per-role MQTT connectivity. A source in `error` MUST flip the top-level status to
`degraded`.
### Self-description {#connector-info}
`GET /admin/connector/info` MUST return the connector identity (`connector_id`), version,
supported `auth_modes`, the per-kind `source_kinds[]` config schemas, the `egress_kinds`
and named `destinations[]`, and additive operational metadata (token expiry, mTLS cert
expiry, config hash, build info). It MUST NOT return any secret value. This is the only
route that MUST be reachable without authentication.
### Error envelope {#management-errors}
Errors MUST be JSON of the shape `{ "error": { "code", "message", "details" } }`, never
plain text. The `code` MUST come from a controlled vocabulary including
`validation_error`, `source_not_found`, `name_conflict`, `state_conflict`,
`auth_required`, `auth_invalid`, `auth_forbidden`, `mtls_subject_not_allowed`,
`auth_not_configured`, `timeout`, `missing_env` and `internal_error`.
## Tasking Control Plane {#tasking}
This section is normative for the **Tasking Control Plane** class. It implements the
connector side of the [[STA-SERVER]] tasking subset and the [[OPENCITYSENSE]] device
lifecycle.
### Keypair and identity {#tasking-keypair}
A connector implementing this class MUST load a libsodium `crypto_box` secret key
(32 bytes, base64) from an environment variable (default
`STA_CONNECTOR_PRIVATE_KEY`). On startup it MUST publish the corresponding public key and
its SHA-256 fingerprint to its own connector [=Thing=] via
`PATCH /v1.1/Things({connector_thing_id})` so that clients can seal [=device secret=]s to
it. The connector Thing's identity MUST be configurable (by id or by name filter).
### Task subscription {#tasking-subscription}
The connector MUST subscribe to the server task topic `v1.1/Tasks/+` over [[MQTT]] and
MUST filter client-side to tasks targeting its own tasking capabilities. The connector
MUST also provide a polling fallback (`GET /v1.1/Tasks?$filter=status eq 'pending'`, by
default every 30 s) so that onboarding still works when the broker is unavailable.
Task handling MUST be idempotent via task-id de-duplication and status transitions.
### Task handlers {#tasking-handlers}
The connector MUST dispatch on `taskingParameters.action` and MUST implement the four
handlers. Each handler MUST transition the Task to `running` when it begins, and to
`completed` or `failed` (with a `result`) when it ends.
- **OnboardDevice** — MUST set the device Thing's controlling connector to itself, MUST
materialise a Datastream per device-model sensor (idempotently), and MUST activate the
device's Configuration.
- **OffboardDevice** — MUST deactivate the active Configuration and clear the device's
controlling-connector link.
- **UpdateConfiguration** — MUST deactivate the current Configuration and activate the
new one from the task payload.
- **RotateSecret** — MUST decrypt the referenced [=device secret=], re-seal it (to its
own public key) and update it.
LoRaWAN network-server registration is out of scope and is a documented no-op in this
revision.
### Device secret handling {#tasking-secrets}
The connector MUST be the only party able to decrypt a [=device secret=]: it MUST decrypt
ciphertext sealed to its public key using a libsodium sealed box ([[NACL]]) with its
private key, and MUST NOT expose decrypted secret material through any interface,
including logs ().
## Authentication & security {#security}
- **Admin network boundary.** The admin surface MUST bind to loopback or a private
interface only. Internet-facing exposure MUST NOT be supported and MUST emit a startup
error.
- **Admin authentication.** Write access MUST require either a bearer token
(`CONNECTOR_ADMIN_TOKEN`, compared in constant time) or mutual TLS
(CA / cert / key, with an allowed-subjects list). `GET /admin/connector/info` is the
only route that MAY be served without authentication, and it MUST NOT return secrets.
- **Egress authentication.** Observations MUST be published with a bearer token resolved
from the environment (), transmitted only over TLS in production.
- **Secrets in environment only.** Tokens, broker passwords and the libsodium private key
MUST be supplied via environment variables and MUST NOT be written to the configuration
file. A missing required variable MUST surface as a source `error`, not a silent
failure.
- **Log redaction.** Stored `log_excerpt`s and emitted logs MUST be passed through secret
redaction so that tokens, passwords and environment values never appear.
- **Quarantine confidentiality.** Raw measurement payloads MUST NOT be exposed through any
admin response.
- **Audit.** Every write to the management API MUST emit an audit log line carrying the
caller IP (or mTLS subject), action, source and result.
## References
Key normative references: [[OGC-STA-1]], [[OGC-STA-2]], [[STA-SERVER]], [[RFC2119]],
[[RFC8174]], [[RFC3339]], [[NACL]], [[MQTT]], [[MODBUS]]. Key informative references:
[[GEONOVUM-TESTBED]], [[OPENCITYSENSE]], [[FROST]] (the reference STA implementation),
[[AIP-136]].