app/docs/reference/connector-contract.md

# The connector contract

A connector is the device-facing node class. It talks to something outside the
engine — a device, a service, a protocol — and publishes what it finds as
ordinary typed messages, so the rest of a flow cannot tell the difference
between a reading from a heat pump and one from a `print` statement.

This document is normative: everything here is what the engine relies on and
what a connector may rely on in return. The authoring walkthrough is in
[Writing a connector](../code/connectors.md).

## Contract version

```python
from fluksio.flow.connector import CONTRACT_VERSION, ConnectorNode

class MyConnector(ConnectorNode):
    contract = CONTRACT_VERSION
```

`contract` must be declared **on the connector class itself**. The loader reads
it from the class `__dict__` rather than through inheritance, so a connector
written against an older base cannot pass itself off as current. A mismatch is
logged and the connector is skipped; nothing else in the engine changes.

The version is bumped only when a change would break connectors written against
the old surface. Additions that leave existing connectors working do not bump
it.

## Packaging and discovery

A connector is an ordinary Python package. It advertises its node class through
an entry point in the `fluksio.node_types` group:

```toml
[project]
name = "fluksio-connector-skeleton"
version = "0.1.0"

[project.entry-points."fluksio.node_types"]
random_sensor = "fluksio_connector_skeleton:RandomSensor"
```

- The **entry point name** is the node type as it appears on the canvas and in
  a stored flow. It must not collide with a built-in type (`python`, `mqtt`,
  `http`, `influxdb`, `delay`, `mlp`); a collision is logged and skipped.
- The **package name and version** are the manifest. There is no second
  metadata file to keep in step, and the editor shows the provenance
  (`fluksio-connector-skeleton 0.1.0`) on the node type.

Discovery happens once, when the engine starts. A connector's code is imported,
and Python does not re-import a changed module, so installing or upgrading one
means restarting the engine.

## Declaring what a connector is

```python
class RandomSensor(ConnectorNode):
    contract = CONTRACT_VERSION
    title = "Random sensor"          # shown in the add-node palette
    description = "Emits a random reading."
```

`title` falls back to the entry point name; `description` may be empty.

## Parameters

Settings are a pydantic model, and the editor builds a form from its JSON
schema. Subclass the base so `poll_interval` stays available:

```python
class Params(ConnectorNode.Params):
    host: str = "localhost"
    api_key: str | None = Field(default=None, json_schema_extra={"x-secret": True})
```

- Booleans render as switches, numbers and strings as inputs. Objects and
  arrays are not rendered — keep settings flat.
- A field carrying **`x-secret`** renders as a picker over the stored secrets
  and writes a reference, `{"$secret": "name"}`, rather than the value. The
  engine resolves it when the node is built, so a credential never lands in
  `flow.json`. Always mark credentials this way.
- The parsed model is available as `self.config`.

## Ports and messages

Ports are declared by whoever places the node, not by the connector, and they
are `MessageSpec`s like any other node's: a message name, a `DType`, and an
optional `interval`.

- `poll()` returns a dict **keyed by output port**, and the base class maps it
  onto message names; `write()` receives a dict **keyed by input port** the
  same way.
- Every value is checked against the port's declared `DType` on the way out. A
  wrong type is an error, not a hint.
- `interval` on a port is enforced by the engine, not by the connector: an
  output port publishes at most every *n* seconds, an input port wakes its node
  at most that often. A connector should poll at the rate the device is
  comfortable with and leave delivery rates to whoever wires it up.
- A key no port declares is an error, and it fails the whole reading rather
  than the one value — a mistyped metric name is how a training curve goes
  missing.

### When the device decides what the ports are

A connector for a device whose readings vary by model — which components a
relay has, which entities were flashed onto a board — cannot know its ports in
advance, and returning everything the device reports would fail on the first
value nobody bound. Narrow the reading to the ports that were declared:

```python
declared = {spec.port for spec in self.output_ports if spec.name}
return {port: value for port, value in reading.items() if port in declared}
```

`spec.port` is the local, unqualified name and stays that way for the node's
whole life, so the set can be taken fresh each time. Say something in the log
when a declared port is not one the device has — once, not once per poll: from
the canvas a renamed entity and a typo look the same, and both leave a port
silent forever.

## Polling

Set `poll_interval` and implement `poll()`; the base class runs the loop:

```python
async def poll(self) -> dict[str, Any] | None:
    return {"reading": await self._read_device()}
```

- Return `None` when there is nothing new.
- **Only changed values are published.** A device polled every few seconds
  usually says the same thing, and every publication wakes everything
  downstream, so the loop compares against what it last published.
- Raising is not fatal: it is reported as a health problem and retried on the
  next tick.
- The loop calls `inject`, which runs the graph, on a worker thread. `poll()`
  itself runs on the event loop and must not block it.

A connector that is pushed to rather than polled leaves `poll_interval` at 0,
overrides `start`/`stop` to open and close its own subscription, and calls
`self.inject({...})` when something arrives.

## Writing

A connector that commands something rather than only reading it implements
`write()`. It is called when the node's inputs are satisfied, with the values
keyed by input port:

```python
def write(self, **ports: Any) -> dict[str, Any] | None:
    self._device.set(ports["level"])
    return None
```

- Settings are already parsed on `self.config`; `write()` is handed ports only.
- Return `None` unless the device answers something worth publishing, in which
  case return it keyed by output port, exactly as `poll()` does.
- It runs on a worker thread rather than the event loop, but the thread is the
  scheduler's, so a long block holds up the cascade. Queue the work if the
  device is slow.
- A connector that only reads leaves it alone.
- `poll()` and `write()` are independent: a connector may have both, and a
  node with inputs and no `poll_interval` never polls.

A write is a command, not a value: set `idempotent = False` on the class so a
redelivery after a crash does not undo a newer command that already landed.

## Devices whose readings are bytes

A camera frame or a recorded clip is far too big to be a message, so a
connector publishes a reference to it instead:

```python
async def poll(self) -> dict[str, Any] | None:
    jpeg = await asyncio.to_thread(self._grab)
    return {
        "frame": self.save_artifact(jpeg, "frame.jpg", media_type="image/jpeg")
    }
```

`save_artifact` stores the bytes and returns what an `image`, `audio` or
`video` port carries — the media type has to match the port's type. It only
works once the node has started, since the store is the engine's and is handed
over then.

Each reading is a new artifact, which the poll loop publishes because its
digest differs from the last. Set `poll_interval` to what somebody actually
wants to look at: a frame every second or two is a glance, and live video
belongs on the camera's own stream rather than in the graph.

## Lifecycle

```python
async def start(self, app=None) -> None: ...
async def stop(self, app=None) -> None: ...
```

- `start` is called when the connector's flow starts, and after every rebuild.
- `stop` is called before a rebuild and when the flow is stopped. **It must be
  idempotent** — it is called whether or not `start` succeeded.
- Overriding either means calling `super()` if the polling loop is also wanted.
- The `app` argument is the FastAPI application, for the rare connector that
  needs to mount a route. Most ignore it.

A stopped flow gets no `start` at all: that is what stopping it means.

## Health

```python
self.report_health("ok")
self.report_health("degraded", "3 of 5 registers timed out")
self.report_health("down", str(exc))
```

Three values, `ok`, `degraded` and `down`, plus an optional detail string. The
engine forwards changes to the editor, which shows them on the node. Reporting
the same status twice is free — only changes are published. The polling loop
already reports around `poll()`; a connector managing its own connection should
report when it connects and when it loses the connection.

Health is about the connection, not about loading: a connector that fails to
load is reported separately and never runs.

## What a connector must not do

- **Hold state that belongs in the flow.** Logic nodes are stateless and run in
  parallel; a connector may hold a connection, a session, or a device handle,
  but a value another node needs belongs in a message.
- **Block the event loop.** `poll`, `start` and `stop` are async and run on the
  loop. Use `asyncio.to_thread` for blocking I/O.
- **Read credentials from the environment.** They come through parameters, so
  the person deploying a flow can change them without touching the host.