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.

pip install cosalette[schema]

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: "1"  # (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

operations:
  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.

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 one open, untyped passthrough field — extra on ha_discovery(), channel_params on openhab() — for platform keys the curated fields don't reach; both 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). merge() raises ValueError if two blocks carry the same extension key. See cosalette ai help consumer-overrides and ADR-056.

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: "1"

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

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.

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.

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 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}.

Further Reading¶

• ADR-033 — MQTT Schema Enforcement — decision record: format choice, distribution model, enforcement modes.
• Reference Network Schema — annotated three-app fleet schema.