Schema Enforcement¶

cosalette apps publish to predictable MQTT topics by construction — the framework sets {prefix}/{device}/state, {prefix}/{device}/set, and {prefix}/{device}/availability automatically. But the payload shape and the cross-app contract (which topics a fleet expects to exist) are not enforced.

Schema enforcement fills that gap using AsyncAPI 3.0.0 documents annotated with x-cosalette-* extensions. The key benefits:

• Catch regressions before deployment. A renamed field in one app silently breaks consumers in another. A schema validation step in your CI/CD pipeline or deploy scripts catches it before the app reaches the broker.
• Machine-readable contract. Monitoring tools, code generators, and dashboards can discover which topics your fleet produces and what payloads to expect.
• Zero friction when unused. The default enforcement mode is off — no new dependencies, no new topics, no broker configuration.

Which Extra Do I Need?¶

cosalette keeps optional features behind extras so a container image installs only what the app uses. This table covers schema, settings-file and MCP dependencies; most apps need no optional schema dependencies at runtime.

Runtime discovery builds its payloads from the app's in-memory AsyncAPI document (ADR-059), so it never reads a schema file. At startup, schema enforcement checks only what its configuration uses (ADR-033 amendment): a YAML schema file needs PyYAML, and jsonschema is needed only when x-cosalette-enforcement.on_publish is true (default false). A schema file whose name ends in .json is parsed with orjson, a core dependency, so it never needs PyYAML; it still needs jsonschema when on_publish is true. A missing dependency fails startup, before MQTT connects, with a dependency hint. A YAML schema with on_publish: true needs both packages, so install cosalette[schema]:

pip install cosalette[schema]

The schema CLI never imports jsonschema; that dependency is used only for publish-time payload validation. Commands that read a .json schema and write JSON or text need neither optional package. PyYAML is needed when a command reads a YAML schema or emits YAML. dump emits YAML by default; dump --format json writes the same document as JSON without PyYAML. init and slice always emit YAML, and ha-discovery --format yaml does too.

A container image without PyYAML can still generate and use a schema:

cosalette schema dump --app myapp.app:app --format json > schema.json
cosalette schema openhab schema.json

JSON has no NaN or Infinity, so a non-finite float in the document (for example a float("inf") bound) is written as null in JSON output. A document key that is not a string cannot be written as JSON at all; dump --format json then exits with a configuration error and suggests --format yaml.

A missing optional dependency is reported with a message and a hint that names the package first, then the extras that contain it, and the JSON alternative where one exists:

Error: YAML output requires pyyaml, which is not installed.

Hint: add pyyaml (or the cosalette[schema] or cosalette[config-yaml] extra) to your project dependencies, or use --format json.

TOML and JSON settings files need no extra (see Configuration — Supported Formats). For the MCP server, see MCP Server.

Quick Start¶

1 — Generate a starter schema from your app¶

# Scaffold a schema with x-cosalette extensions included
cosalette schema init --app myapp.app:app > schema.yaml

init introspects the running app via app.asyncapi() and produces a canonical AsyncAPI 3.0.0 document, then layers on an x-cosalette-enforcement scaffold:

asyncapi: 3.0.0
info:
  title: thermo2mqtt
  version: 0.1.0
  x-cosalette-contract-version: "2"  # (1)!

x-cosalette-enforcement:             # (2)!
  mode: warn
  on_configure: true
  on_publish: false
  network_level: false

channels:
  temperatureState:
    address: thermo2mqtt/temperature/state
    x-cosalette-app: thermo2mqtt       # (3)!
    x-cosalette-archetype: telemetry
    messages:
      message:
        payload:
          type: object  # (4)!

  setpointCommand:
    address: thermo2mqtt/setpoint/set
    x-cosalette-app: thermo2mqtt
    x-cosalette-archetype: command
    messages:
      message:
        payload:
          type: object

  temperatureAvailability:           # (5)!
    address: thermo2mqtt/temperature/availability
    x-cosalette-app: thermo2mqtt
    messages:
      message:
        payload:
          type: string
          enum: [online, offline]
    bindings:
      mqtt: {qos: 1, retain: true}
    x-cosalette-framework: availability
    x-cosalette-discoverable: false

  # setpointAvailability: same shape, at thermo2mqtt/setpoint/availability

operations:
  publishTemperatureAvailability:
    action: send
    channel:
      $ref: '#/channels/temperatureAvailability'
  publishTemperatureState:
    action: send
    channel:
      $ref: '#/channels/temperatureState'
  receiveSetpointCommand:
    action: receive
    channel:
      $ref: '#/channels/setpointCommand'

1. Framework-managed contract-shape version — see Contract Version Metadata below.
2. Enforcement scaffold added by init for editing; absent from dump output.
3. App-ownership tag emitted from the App registry on every channel — survives regeneration, so downstream consumers (e.g. schema ha-discovery) resolve the owning app without hand-editing.
4. Fallback when no state_model or typed return annotation is registered. Add properties and constraints by hand, or declare state_model on the decorator. @app.telemetry, @app.command, and @app.device all accept state_model= to type their state channels.
5. Framework availability channel, one per device, telemetry, command and named stream (contract version "2", ADR-086). It has no archetype, is never a Home Assistant / openHAB entity and adds no ACL grant.

2 — Add payload constraints¶

Edit the generated YAML to specify required fields and types:

channels:
  temperatureState:
    address: thermo2mqtt/temperature/state
    x-cosalette-app: thermo2mqtt
    x-cosalette-archetype: telemetry
    messages:
      reading:
        payload:
          type: object
          required: [temperature, unit]
          properties:
            temperature:
              type: number
              x-cosalette-consumer:
                device_class: temperature
                unit: "°C"
                display_name: Room Temperature
                state_class: measurement
            unit:
              type: string
              enum: [celsius, fahrenheit]

The x-cosalette-consumer annotation carries metadata for downstream consumer code generation (e.g. home automation integrations, dashboard provisioning). It is optional — omit it if you do not need consumer code generation.

Producing the block: the consumer() helper¶

When you author payloads as pydantic models (rather than hand-writing YAML), attach the block with the typed consumer() producer instead of a hand-built dict:

from typing import Annotated

import pydantic
from cosalette.schema import consumer


class TemperatureReading(pydantic.BaseModel):
    temperature: Annotated[
        float,
        pydantic.Field(
            json_schema_extra=consumer(
                device_class="temperature",
                unit="°C",
                display_name="Room Temperature",
                state_class="measurement",
            )
        ),
    ]

consumer(**meta) returns {"x-cosalette-consumer": {...}} ready to pass to Field(json_schema_extra=...). Its keyword arguments are typo-checked under a type checker (ty/pyright) at author time against ConsumerMeta, whose key set (display_name, device_class, unit, state_class, icon, read_only) is the single source of truth shared with the ConsumerMetadata reader. This is a static check only: at runtime the reader silently ignores unknown keys. The block rides on the field, so it survives schema regeneration via TypeAdapter(model).json_schema() and feeds the Home Assistant / OpenHAB discovery generators. See cosalette ai help consumer.

Semantic presets: temperature() and percent()¶

Many fields share a fixed set of metadata keys, differing only in the display_name. cosalette.schema.temperature() and cosalette.schema.percent() wrap consumer() for the two most common shapes:

from typing import Annotated

import pydantic
from cosalette.schema import percent, temperature


class BoilerState(pydantic.BaseModel):
    flow_temperature: Annotated[
        float, pydantic.Field(json_schema_extra=temperature("Flow Temperature"))
    ]
    modulation: Annotated[int, pydantic.Field(json_schema_extra=percent("Modulation"))]
    pump_speed: Annotated[
        int,
        pydantic.Field(json_schema_extra=percent("Pump Speed", icon="mdi:pump")),
    ]

temperature(display_name) returns consumer(display_name=..., device_class="temperature", unit="°C", state_class="measurement"). percent(display_name, *, icon=None) returns consumer(display_name=..., unit="%", state_class="measurement"), adding icon only when supplied — omitted, not None, when left out — so the output matches a hand-written block exactly.

Platform overrides: ha_discovery() and openhab()¶

x-cosalette-ha-discovery and x-cosalette-openhab (below) have their own typed producers, mirroring consumer()'s pattern. Because a field usually needs consumer() plus one or both override producers, combine them with merge() — Field(json_schema_extra=...) accepts only one dict:

from typing import Annotated

import pydantic
from cosalette.schema import consumer, ha_discovery, merge, openhab


class BulbState(pydantic.BaseModel):
    hsb: Annotated[
        list[int],
        pydantic.Field(
            json_schema_extra=merge(
                consumer(display_name="HSB"),
                openhab(
                    item_type="Color",
                    channel_type="color",
                    channel_params={"colorMode": "HSB"},
                ),
            )
        ),
    ]
    state: Annotated[
        bool,
        pydantic.Field(
            json_schema_extra=merge(
                consumer(display_name="Desk Lamp"),
                ha_discovery(extra={"schema": "json", "optimistic": False}),
            )
        ),
    ]

ha_discovery(**meta) and openhab(**meta) are typo-checked against HaDiscoveryMeta / OpenHabMeta, the single source of truth shared with the HaDiscoveryOverrides / OpenHabOverrides readers. Each also carries open, untyped passthrough fields — extra on ha_discovery(), channel_params and thing_params on openhab() — for platform keys the curated fields don't reach; all are merged in last, so they can add a new key or override a computed default (e.g. the switch channel's default on/off, or the Thing's computed availabilityTopic). merge() raises ValueError if two blocks carry the same extension key. See cosalette ai help consumer-overrides, ADR-056 and ADR-079.

3 — Validate the schema document¶

cosalette schema validate schema.yaml

✅ Schema validated: thermo2mqtt v0.1.0
   AsyncAPI version: 3.0.0
   Channels: 2
   Schema type: single-app

This checks structure and cosalette-specific extension syntax. It does not run the app — it validates the schema file alone.

4 — Check app registrations against the schema¶

cosalette schema check --app thermo2mqtt.app:app --schema schema.yaml

Schema: schema.yaml (v0.1.0)
App:    thermo2mqtt

✓ temperature — OK
✓ setpoint — OK

Result: 0 violations, 2 compliant
Exit code: 0

If a device is missing or a scope rule is violated, check exits with code 1 and prints the violation:

✗ setpoint — MISSING
    Schema expects device 'setpoint' but no registration found

Result: 1 violations, 1 compliant
Exit code: 1

5 — Enable enforcement in the app¶

Schema enforcement is configured through the framework's nested settings model under the schema key. The two relevant fields are:

Set them in your .env file (or environment):

SCHEMA__PATH=/etc/cosalette/thermo2mqtt-schema.yaml
SCHEMA__ENFORCEMENT=warn

Or in a settings subclass if you prefer code-level defaults:

from cosalette import App, Settings
from cosalette._settings import SchemaSettings


class MySettings(Settings):
    schema_: SchemaSettings = SchemaSettings(
        path="/etc/cosalette/thermo2mqtt-schema.yaml",
        enforcement="warn",
    )


app = App(name="thermo2mqtt", settings_class=MySettings)

Enforcement Modes¶

Configure in the schema file:

x-cosalette-enforcement:
  mode: warn          # off | warn | strict
  on_configure: true  # validate device registrations at startup
  on_publish: false   # validate payload shape at publish time (dev only)
  network_level: false

Network-Level Schema¶

For a fleet of multiple cosalette apps, a network-level schema defines the entire MQTT topology in one file. Each app validates against its own slice.

When to use it¶

A network schema is the primary use case for cross-app validation:

• One app renames a topic → the network schema flags it; the deploy is blocked before the change reaches the broker.
• An app adds a new channel → cosalette schema check reports it as "extra" (not a violation) so you can decide whether to promote it to the schema.
• A CI/CD gate validates all apps in a single step before any of them are deployed.

Network schema structure¶

asyncapi: 3.0.0
info:
  title: My MQTT Network
  version: 1.0.0

x-cosalette-enforcement:
  mode: warn
  network_level: true   # marks this as a network-level schema

channels:
  thermoTemperatureState:
    address: thermo2mqtt/temperature/state
    x-cosalette-app: thermo2mqtt           # channel belongs to this app
    x-cosalette-archetype: telemetry
    messages:
      reading:
        payload:
          type: object
          required: [temperature]
          properties:
            temperature:
              type: number

  airsenseAirQualityState:
    address: airsense2mqtt/airquality/state
    x-cosalette-app: airsense2mqtt
    x-cosalette-archetype: telemetry
    messages:
      reading:
        payload:
          type: object
          required: [co2, humidity]
          properties:
            co2:
              type: integer
            humidity:
              type: number

Extract an app's slice¶

The slice command filters a network schema to a single app's channels:

cosalette schema slice --network /etc/cosalette/network-schema.yaml --app thermo2mqtt

asyncapi: 3.0.0
info:
  title: thermo2mqtt
  version: 2.0.0

x-cosalette-enforcement:
  mode: warn
  on_configure: true
  on_publish: false
  network_level: false

channels:
  thermoTemperatureState:
    address: thermo2mqtt/temperature/state
    x-cosalette-app: thermo2mqtt
    x-cosalette-archetype: telemetry
    ...

You can pipe this directly to a file or use it for per-app validation:

cosalette schema slice \
  --network /etc/cosalette/network-schema.yaml \
  --app thermo2mqtt > /etc/cosalette/thermo2mqtt-schema.yaml

Check against a network schema¶

check automatically filters the network schema to the app's slice:

cosalette schema check \
  --app thermo2mqtt.app:app \
  --schema /etc/cosalette/network-schema.yaml

The command detects network_level: true, extracts the thermo2mqtt slice, and validates the app's registrations against it. No separate slice step needed.

Reference schema¶

See docs/assets/reference-network-schema.yaml for a complete annotated example covering a three-app smart-home fleet (thermo2mqtt, airsense2mqtt, solarray2mqtt) with telemetry, command, and fleet-wide availability channels.

Deployment Integration¶

cosalette schema check is a standard subprocess that exits 0 on compliance and 1 on violations, so it integrates cleanly into any deploy toolchain.

Shell / CI scripts¶

The simplest form — run before starting each app:

cosalette schema check \
  --app myapp.app:app \
  --schema /etc/cosalette/network-schema.yaml || exit 1

# Start the app only if validation passed
myapp start

This works in any environment: bare-metal init scripts, Docker entrypoints, GitHub Actions steps, GitLab CI jobs, or Makefile targets.

GitHub Actions example¶

- name: Validate schema
  run: |
    cosalette schema check \
      --app myapp.app:app \
      --schema schemas/network-schema.yaml

Ansible example¶

For teams using Ansible to manage hosts, the pattern below deploys the schema file first and validates before starting the service:

# tasks/deploy-myapp.yml

- name: Deploy network schema
  ansible.builtin.copy:
    src: files/network-schema.yaml
    dest: /etc/cosalette/network-schema.yaml
    mode: "0644"

- name: Validate myapp against network schema
  ansible.builtin.command:
    cmd: >
      cosalette schema check
        --app myapp.app:app
        --schema /etc/cosalette/network-schema.yaml
  changed_when: false
  failed_when: result.rc != 0
  register: result

- name: Start myapp service
  ansible.builtin.systemd:
    name: myapp
    state: started
  when: result.rc == 0

The exit-code contract is the same regardless of toolchain — check failing blocks the next step.

x-cosalette Extension Reference¶

Channel-level extensions¶

Payload-level extensions (whole payload model)¶

Property-level extensions¶

Document-level enforcement config¶

x-cosalette-enforcement:
  mode: warn          # off | warn | strict
  on_configure: true  # validate at startup
  on_publish: false   # validate payload at publish time
  network_level: false

Info-level metadata¶

Contract Version Metadata¶

x-cosalette-contract-version appears in the info section of every AsyncAPI document generated by app.asyncapi() (and therefore by cosalette schema dump, cosalette schema init, cosalette manifest, and the cosalette_manifest MCP tool):

info:
  title: myapp
  version: 1.2.0
  x-cosalette-contract-version: "2"

This value tracks the shape of the generated document independently from the application version. Its semantics:

Version history:

Migration guidance: When x-cosalette-contract-version increases after a cosalette upgrade, re-run cosalette schema dump to regenerate your committed schema baseline. The CHANGELOG documents every bump with a description of the structural change. Hand-maintained schema files (schema.yaml) are not affected — they are only validated against the running app, not regenerated automatically.

Topic Prefix and Channel Addresses¶

A channel's address is transport: it is composed from the resolved MQTT topic prefix, settings.mqtt.topic_prefix or App(name=...) — the same resolution the runtime performs. A channel's x-cosalette-app (and info.title) is identity: always the app name, never the prefix. The two are never interchangeable (see ADR-072).

There are two ways to make dump / init emit the deployment's real addresses:

# 1. Resolve it from settings (runs the app's configure hooks):
cosalette schema dump --app myapp.app:app --resolve-settings --env-file prod.env

# 2. State it outright (no configure hooks, no settings files):
cosalette schema dump --app myapp.app:app --topic-prefix house/wiz

--topic-prefix exists for CI gates that must not execute application code beyond the import. When both flags are given, the explicit --topic-prefix wins for the prefix, while --resolve-settings still runs the configure/expand lifecycle so ADR-023 callable name= registrations are expanded. Multi-segment prefixes (house/wiz) are supported; MQTT wildcards are rejected. The prefix changes addresses only — identity, and therefore Home Assistant object_id / unique_id, is unaffected. To run several instances of one app, see Several instances of one app.

Generating the Document from Python¶

cosalette.schema.resolved_asyncapi() is an async function that returns the document cosalette schema dump --resolve-settings writes, as a dict. Use it in tests and build scripts that already import the app, instead of running the CLI and parsing its output:

from cosalette.schema import (
    SchemaBuildError,
    resolved_asyncapi,
    resolved_asyncapi_sync,
)

from myapp.app import app

doc = await resolved_asyncapi(app, env_file="prod.env")
# Same as: cosalette schema dump --app myapp.app:app --resolve-settings \
#          --env-file prod.env --format json

For a script that has no running event loop, call resolved_asyncapi_sync(app, ...) with the same keyword arguments.

It takes the inputs of dump --resolve-settings: env_file= (--env-file), config_file= (--config-file) and topic_prefix= (--topic-prefix). For the same app and inputs the dict equals the parsed --format json output. Without env_file=, a .env file in the working directory is read if it exists, as the CLI does.

The app is not modified. Settings-derived (ADR-023) name= and topic= specs are expanded on copies of its registrations, so app.asyncapi() still describes the unresolved app and repeated calls return equal documents. Adapter factories and on_configure hooks do run, since resolution depends on them; adapters are resolved in dry-run mode, and no store or adapter lifecycle is entered.

Failures are raised, not printed:

An exception raised by an adapter factory or a configure hook propagates unchanged. The CLI stays the primary interface: the function follows it (ADR-051).

CLI Reference¶

Consumer Code Generation¶

Properties annotated with x-cosalette-consumer can be transformed into consumer platform configurations automatically. This eliminates hand-maintaining discovery payloads and configuration files — the AsyncAPI schema becomes the single source of truth.

Home Assistant MQTT Discovery¶

Generate HA discovery payloads that Home Assistant accepts via its MQTT discovery protocol:

cosalette schema ha-discovery network.yaml

Output is a JSON array of {topic, config} objects — one per annotated property. Each object contains the discovery topic and the full config payload. Publish these as retained messages and HA will auto-create entities.

Component inference: When x-cosalette-ha-discovery.component is not set, the component is inferred from archetype and JSON schema type:

Optional fields (int | None, emitted as anyOf: [T, null]) infer from their non-null variant T rather than degrading to sensor/string.

Command entities publish a JSON envelope. cosalette's wire format is one JSON object per channel, so a command entity gets a command_template of {"<prop>": {{ value }}} (string properties are quoted) unless you supply an explicit command_template. Command entities also carry a _cmd suffix in their object_id / unique_id / discovery topic so a state entity and a command entity for the same device and property never collide. State entities keep their bare identifiers, so existing read-only apps are unaffected.

Schema constraints are honoured. number entities emit min / max / step from minimum / maximum / multipleOf; select entities emit options from enum; array properties render via a join filter ({{ value_json.rgb | join(',') }}). Keys the target platform rejects are dropped — a binary_sensor does not receive unit_of_measurement or state_class.

read_only forces a read-only component (binary_sensor / sensor) and emits a state_topic only, never a command_topic.

Composite entities¶

The overrides above still produce one HA entity per property. For a single entity spanning several JSON fields — a light with state + brightness + color_temp, a climate, a cover with position — declare it on the payload model, not a field, with ha_entities(ha_entity(...)):

from typing import Annotated
import pydantic
from cosalette.schema import consumer, ha_entities, ha_entity


class BulbState(pydantic.BaseModel):
    model_config = pydantic.ConfigDict(
        json_schema_extra=ha_entities(
            ha_entity(
                component="light",
                name="Desk Lamp",
                extra={
                    "schema": "json",
                    "brightness": True,
                    "supported_color_modes": ["color_temp", "hs"],
                },
            ),
        )
    )

    state: Annotated[bool, pydantic.Field(json_schema_extra=consumer())]
    brightness: Annotated[int, pydantic.Field(json_schema_extra=consumer())]

A channel whose payload model declares ha_entities skips per-property scalar generation entirely for that channel — the composite entity replaces the scatter, it does not add to it.

component selects a real payload builder, not just a topic segment:

extra is merged last, same override-last semantics as ha_discovery().extra.

A device archetype channel with payload_model= emits a paired /state (send) and /set (receive) channel sharing one model — the entity's two topic halves are merged into one config automatically rather than emitted twice, each incomplete.

To surface a list or object payload as Home Assistant attributes, set json_attributes_template in extra. cosalette then defaults json_attributes_topic to the entity's own state topic (ADR-075) — HA ignores the template without a topic, and because the default is the channel's resolved address, a model-level spec shared by callable-named (name=) channels names each channel's own topic instead of hard-coding one. An explicit json_attributes_topic in extra still wins. HA requires the template to render a JSON object (attribute-name → value), so a bare list must be wrapped under a key. This lets a list payload expose its contents as attributes, not just a count:

ha_entity(
    component="sensor",
    name="birthday",
    extra={
        "value_template": "{{ value_json.events | length }}",
        "json_attributes_template": "{{ {'events': value_json.events} | tojson }}",
    },
)  # json_attributes_topic defaults to the channel's /state address

See ADR-057 for the full design rationale.

Example:

# In your AsyncAPI schema
temperature:
  type: number
  x-cosalette-consumer:
    device_class: temperature
    unit: '°C'
    display_name: 'Heating Water Temperature'
    state_class: measurement
  x-cosalette-ha-discovery:
    expire_after: 300

Produces a discovery payload at homeassistant/sensor/<app>/<device>_temperature/config with device_class, unit_of_measurement, state_class, value_template, and expire_after fields set correctly.

Availability and device modelling¶

Every generated HA entity (scalar or composite) carries availability, an origin block, and a device block modelled on the resolved device, not the app:

• Availability mirrors the ADR-012 rule the runtime HealthReporter uses. A resolved device gets availability (a list combining its own {app}/{device}/availability topic with the app-level {app}/status heartbeat/LWT topic) and availability_mode: "all", so the entity shows unavailable if either the device or the app process goes down — closing the gap where an app crash only flips {app}/status, leaving per-device topics stuck at a stale "online". An app with no named devices (a single-device app publishing directly to {app}/state) gets the simpler availability_topic: "{app}/availability" instead.
• device is one HA device per resolved device (identifiers: ["cosalette_<app>_<device>"]), linked to an app-level bridge device via via_device, instead of collapsing every entity in an app into a single HA device.
• origin ({name: <app>, sw_version: <app_version>}) is added to every payload for free — HA surfaces it in entity diagnostics.
• A synthetic diagnostic binary_sensor (device_class: connectivity, sourced from {app}/status) is emitted once per app that has at least one named device. HA's device registry does not create a device purely from a via_device reference — this entity is what makes the bridge device (and therefore via_device) actually appear in Home Assistant.

See ADR-058 for the full design rationale.

openHAB gets the same signal through a different mechanism — see OpenHAB availability below.

Excluding a channel from discovery¶

Not every channel is a consumer entity. A diagnostic counter or an internal event feed published as telemetry has no place in Home Assistant. Declare that intent at registration with discoverable=False:

@app.telemetry("diagnostics", interval=60, discoverable=False)
async def diagnostics() -> dict:
    return {"loop_lag_ms": measure()}

The channel is then excluded from schema ha-discovery / schema openhab output and emits x-cosalette-discoverable: false on its generated channel — only when set, so a document with no opt-outs is byte-identical to one generated before ADR-073. This is the durable, author-controlled replacement for hand-editing the archetype, which schema dump erases on the next regeneration.

A @app.command with both payload_model and state_model — or a @app.device with payload_model — emits two channels: a /set command channel and a /state channel. discoverable=False opts out both. To opt out only one, pass discoverable="state" (keep the /state channel, hide the /set command channel) or discoverable="command" (the reverse). This is the correct shape for a command whose paired read-only state is an entity while the command itself is intentionally not — before ADR-074, discoverable=False deleted the working state entity along with the command:

@app.command(
    "display",
    payload_model=DisplayCommand,
    state_model=DisplayState,
    discoverable="state",
)
async def display(payload: DisplayCommand) -> DisplayState: ...

The literal resolves to a per-channel boolean at generation time, so the emitted x-cosalette-discoverable stays a plain boolean and the loader round-trip is unchanged.

The discovery gate is evaluated per channel: schema ha-discovery / schema openhab exit non-zero and name every consumer-visible channel that produces no entity — one annotated channel no longer covers for an un-annotated sibling. The message distinguishes three causes: annotations that were present but skipped (array items, or a top-level array-of-objects — for Home Assistant, declare a channel-level ha_entities() composite); a composite that is declared but which the target does not render (openHAB only — see the warning above); and channels with no annotations at all (add consumer()/ha_entities(), or mark the channel discoverable=False).

A top-level array-of-objects property (events: list[Event]) produces no scalar entity: like an array item, the array itself has no single value, so a value_json.events | join(',') render would be an invalid Python repr that Home Assistant drops once the list crosses its 255-character state limit. An array of scalars still renders join(',').

Typed value aggregates (ADR-076). The durable, cross-target remedy is to give the array a single value with consumer(aggregate=...). From one declaration naming no target, each generator renders in its own vocabulary — openHAB transformationPattern="JSONPATH:$.events.length()", Home Assistant value_template: {{ value_json.events | length }}. count is valid on any array; min/max/avg/sum reduce an array of numbers and are rejected at generation time on a non-numeric array. A bare count leaves unit unset, so openHAB resolves it to a DecimalType Number. On openHAB a missing key discards the message and the Item keeps its previous value rather than going UNDEF; an empty array yields 0. Reach for a channel-level ha_entities() composite only when you need the list contents as Home Assistant attributes — which serves Home Assistant only; schema openhab has no composite equivalent.

Several instances of one app¶

Discovery ids derive from the app name: the discovery topic homeassistant/<component>/<app>/<object_id>/config, unique_id cosalette_<app>_<object_id>, and the device identifiers and via_device. Two instances of the same app on one broker therefore claim the same entities, even when their MQTT__TOPIC_PREFIX differs: the last one to publish wins and Home Assistant shows only one instance.

Give each instance its own identity with MQTT__INSTANCE_ID (ADR-089):

# docker-compose.yml
services:
  wiz-attic:
    environment:
      MQTT__TOPIC_PREFIX: house/attic/wiz
      MQTT__INSTANCE_ID: wiz_attic
  wiz-cellar:
    environment:
      MQTT__TOPIC_PREFIX: house/cellar/wiz
      MQTT__INSTANCE_ID: wiz_cellar

The instance id replaces the app name in every identity field: node_id, unique_id, device identifiers, via_device, the bridge device name, and the store key that discovery orphan cleanup uses, so one instance never clears another's configs even when both share a store. In openHAB output it names the Thing UID, Thing label, Item ids and item group. Topics still follow MQTT__TOPIC_PREFIX, and the origin block still names the app. The offline generators take the same value:

cosalette schema ha-discovery schema.yaml --instance-id wiz_attic
cosalette schema openhab schema.yaml --instance-id wiz_attic

--instance-id is rejected for a document that describes several apps. Ids start with a lowercase letter and contain lowercase letters, digits, and single underscores between nonempty segments ([a-z][a-z0-9]*(?:_[a-z0-9]+)*). Uppercase, hyphens, leading digits, repeated underscores, and trailing underscores are rejected instead of being normalized into another instance's identity. OpenHAB Item and group prefixes preserve these underscores; unset ids retain the existing CamelCase output. Flat identities use a length-prefixed token, i<length>_<id> (for example i9_wiz_attic), before appending device/entity segments. This keeps instance a with device b_sensor distinct from instance a_b with device sensor. The token appears in HA unique_id, bridge/device identifiers, via_device, and openHAB Thing UIDs, channel links, Item ids and groups. Discovery topic node_id and display labels still use the raw ID.

When the id is unset, output is byte-identical to earlier releases, so existing entities keep their unique_ids. Setting it on a deployment that is already discovered changes its ids once: Home Assistant creates new entities, and the old retained configs stay until you delete the old device in Home Assistant. With runtime discovery enabled, the app logs a warning at startup when MQTT__TOPIC_PREFIX differs from the app name and no instance id is set, because that is the usual sign of a second instance.

OpenHAB Configuration¶

Generate OpenHAB .things and .items files:

cosalette schema openhab network.yaml --output both
cosalette schema openhab network.yaml --output things
cosalette schema openhab network.yaml --output items

Things use JSONPATH transformations to extract individual properties from JSON payloads. Items are typed according to device_class (e.g. Number:Temperature) or explicit x-cosalette-openhab.item_type overrides.

Each resolved device produces one Thing (the union of its state and command channels) rather than one Thing per channel, so a bidirectional device no longer emits duplicate Thing UIDs. Command channels and their Items carry a _cmd suffix so they never collide with the state side, and both sides derive their names from the same helper so an Item always links to the channel it names. Command channels build an outbound JSON envelope with formatBeforePublish ({"<prop>":%s}) instead of an inbound-only transformationPattern, and boolean switch channels emit on="true" / off="false" so JSON booleans are not left UNDEF. Nested device addresses ({app}/{room}/{device}/state) resolve the device to {room}/{device}.

OpenHAB availability¶

Every Thing carries a Thing-level config bracket wiring the openHAB MQTT binding's availability support to the retained topic the runtime HealthReporter already publishes (ADR-079):

Thing mqtt:topic:broker:wiz2mqtt_desk "wiz2mqtt desk" (mqtt:broker:broker) [
    availabilityTopic="wiz2mqtt/desk/availability",
    payloadAvailable="online",
    payloadNotAvailable="offline"
] {
    Channels:
        ...
}

Named devices point at {prefix}/{device}/availability; an app with no named devices points at {prefix}/availability — the same resolution the HA generator uses. Unlike HA, the openHAB binding accepts a single topic, so the app-level {prefix}/status LWT cannot be combined in: after an unclean crash the retained availability topics stay at "online" and the Thing remains ONLINE until the app restarts. This is a documented trade-off; operators who prefer the crash-accurate signal over per-device granularity can override the computed default with thing_params:

openhab(
    thing_params={
        "availabilityTopic": "wiz2mqtt/status",
        "transformationPattern": "JSONPATH:$.status",
    }
)

thing_params is merged into the Thing bracket last. When several properties of one Thing set it, they merge in channel-address then property-name order and later entries win for the same key.

Further Reading¶

• ADR-033 — MQTT Schema Enforcement — decision record: format choice, distribution model, enforcement modes.
• ADR-079 — OpenHAB Availability Wiring — Thing-level availability bracket, single-topic trade-off, thing_params.
• Reference Network Schema — annotated three-app fleet schema.