Migrate Between cosalette Versions¶

Task-supervision behavior change¶

Device and stream handlers that raise no longer end silently. The framework now reports the failure and, by default, restarts the task with backoff. A task that exhausts its restart budget shuts down the app with CLI exit code 4; configure your process supervisor to restart on that code, or set on_task_failure="ignore" or "exit" to choose a different policy. See Task Supervision.

Telemetry error reminders (v0.11.0+)¶

A telemetry error that persists is no longer reported only once. The framework logs a WARNING and republishes the error on the 2^nd, 4^th, 8^th, ... failure in the first hour of the streak, then once an hour, and the recovery line states how long the outage lasted (ADR-082). Telemetry error payloads now carry details.count and details.first_seen, and heartbeat telemetry entries gain last_error and failing_since. Consumers that treat every message on {prefix}/error as a new incident should read details.count; to keep the old onset-only behavior, pass App(error_reminder_interval=None).

This guide covers breaking changes and mechanical rewrites when upgrading between cosalette versions. For AI-assisted migration from non-cosalette IoT apps, see AI-Assisted Development.

Router Composition (v0.2.0+)¶

When to Adopt Router¶

Router is not Router.include_router — it's single-level composition only.

Mechanical Migration¶

Before — direct app-level registration:

import cosalette

app = cosalette.App(name="home2mqtt", version="1.0.0")


@app.telemetry("temperature", interval=30)
async def read_temperature() -> dict[str, object]:
    return {"celsius": 22.5}

After — Router module with app.include_router():

import cosalette

router = cosalette.Router(prefix="sensors", tags=["environment"])


@router.telemetry("temperature", interval=30)
async def read_temperature() -> dict[str, object]:
    return {"celsius": 22.5}

import cosalette
from sensors import router as sensors_router

app = cosalette.App(name="home2mqtt", version="1.0.0")
app.include_router(sensors_router)

MQTT topic changes:

• Before: home2mqtt/temperature/state
• After: home2mqtt/sensors/temperature/state

The prefix="sensors" parameter adds a topic segment. Omit prefix to keep original topics.

See Router Composition for multi-module organization patterns.

Typed Payloads and Returns (v0.4.0+)¶

Raw String → Pydantic Models¶

Before — raw string payload:

@app.command("valve")
async def handle_valve(payload: str, ctx: cosalette.DeviceContext) -> None:
    import json

    data = json.loads(payload)
    position = data["position"]
    # ... driver logic ...
    await ctx.publish_state(json.dumps({"position": position}))

After — typed payloads with Annotated[T, Payload()]:

from typing import Annotated
from pydantic import BaseModel
from cosalette.mqtt import Payload


class ValveCommand(BaseModel):
    position: int  # 0–100


class ValveState(BaseModel):
    position: int
    flow_lpm: float


@app.command("valve")
async def handle_valve(
    cmd: Annotated[ValveCommand, Payload()],
) -> ValveState:
    # ... driver logic ...
    return ValveState(position=cmd.position, flow_lpm=2.3)

Raw escape hatch — when you need the unmodified string:

# By parameter name convention
async def handler(payload: str) -> dict[str, object]: ...


# Or with explicit marker
async def handler(raw: Annotated[str, Payload(raw=True)]) -> dict[str, object]: ...

See Contract-First Route Design for full patterns.

Triggerable Telemetry with Typed Payloads¶

Before — TriggerPayload data wrapper:

from cosalette.contracts import TriggerPayload


@app.telemetry("sensor", interval=300, triggerable=True)
async def sensor(trigger: TriggerPayload) -> dict[str, object]:
    if trigger.is_triggered:
        days = int(trigger.data or "7")
    else:
        days = 7
    return {"data": await read_sensor(days=days)}

After — Annotated[Model | None, Payload()]:

from typing import Annotated
from pydantic import BaseModel
from cosalette.mqtt import Payload


class RefreshCommand(BaseModel):
    days: int = 7


@app.telemetry("sensor", interval=300, triggerable=True)
async def sensor(
    cmd: Annotated[RefreshCommand | None, Payload()],
) -> dict[str, object]:
    days = cmd.days if cmd is not None else 7
    return {"data": await read_sensor(days=days)}

On scheduled runs, cmd is None. On triggered runs, it holds the validated model.

Trigger Sources (v0.8.0+)¶

triggerable= widened from a bool flag into a trigger-source declaration (ADR-064). True/False still work and mean exactly what they always did, so no application code has to change.

interval= is still required for every triggerable entity — it is the heartbeat and the fallback poll when no trigger arrives.

Breaking change: TelemetryRegistration.triggerable¶

The introspection field is no longer a bool. It is now TriggerSource | None — "mqtt", "local", "both", or None:

# Before (≤ 0.7.x)
if reg.triggerable:  # bool
    ...

# After (0.8.0+)
if reg.triggerable is not None:  # any trigger source
    ...
if reg.triggerable in ("mqtt", "both"):  # subscribes /set
    ...

Truthiness is unchanged for the common case (None is falsy, every source string is truthy), so if reg.triggerable: keeps working; only code that compared against True/False or annotated the field as bool needs updating.

The manifest gained a matching trigger_source field; its existing triggerable field stays a bool and is unchanged.

New: local (in-process) triggers¶

triggerable="local" wakes a telemetry entity from your own code instead of from MQTT, via the new injectable EntityNotifier:

import cosalette
from cosalette import EntityNotifier


@app.state
def bus(notify: EntityNotifier) -> SensorBus:
    return SensorBus(on_reading=lambda: notify("pressure"))


@app.telemetry("pressure", interval=300, triggerable="local")
async def pressure(trigger: cosalette.TriggerPayload) -> dict[str, object]:
    return {"bar": await read_pressure(), "why": trigger.source}

TriggerPayload.source is new and reports "scheduled", "mqtt" or "local". See Local (In-Process) Triggers.

No MQTT, discovery or AsyncAPI output changes. triggerable="local" subscribes nothing, and no generated artifact reads triggerable=.

New: local triggers on @app.device¶

@app.device accepts triggerable="local" too, so a device entity can publish the instant something happens instead of on its next poll. A device owns its own loop, so instead of the framework racing an interval= on its behalf the handler awaits an injected DeviceTrigger (ADR-065):

from cosalette import DeviceContext, DeviceTrigger


@app.device("gadget", triggerable="local")
async def gadget(ctx: DeviceContext, trigger: DeviceTrigger) -> AsyncIterator[None]:
    while True:
        # Wakes on notify("gadget"); the timeout is a heartbeat so the
        # retained state topic still refreshes if the hardware goes quiet.
        await trigger.wait(timeout=60.0)
        await ctx.publish_state(read_gadget())
        yield

Two differences from telemetry are worth noting:

• Devices accept triggerable="local" only. {prefix}/{device}/set is already the device's command topic, so True/"mqtt"/"both" are rejected at registration time — handle those messages with ctx.on_command() instead.
• triggerable= and the DeviceTrigger parameter must agree. Declaring either one without the other raises at registration rather than silently never waking.

This is purely additive: existing @app.device registrations are unaffected.

New: min_interval= storm throttle¶

A push source can wake a handler far faster than the handler is useful. Before 0.8.0 the only remedy was a rate limit at the notify() call site. @app.telemetry and @app.device now accept min_interval= (ADR-066), which bounds the minimum spacing between trigger-initiated run starts:

# Before (≤ 0.7.x) — hand-rolled dedup at the call site
def _on_reading(value: float) -> None:
    if value != _last_value:  # manual guard against a push storm
        notify("pressure")


# After (0.8.0+) — the framework throttles, nothing is dropped
@app.telemetry("pressure", interval=300, triggerable="local", min_interval=2.0)
async def pressure(trigger: cosalette.TriggerPayload) -> dict[str, object]:
    return {"bar": await read_pressure()}

The throttle is leading edge plus trailing edge: the first wake after a quiet period runs immediately, wakes arriving inside the window coalesce into exactly one run when it reopens, and that run carries the last payload. interval= heartbeats are never throttled and never consume a held wake.

min_interval= requires triggerable= and must be a finite, strictly positive number of seconds; both mistakes raise ValueError at registration. The default None leaves existing behaviour byte-for-byte unchanged, so this is purely additive. See Throttling a Trigger Storm.

Automatic Telemetry/Device Availability (v0.9.6+)¶

Breaking. The symptom you will notice first: entities that used to stay available with a stale value now go unavailable in Home Assistant when reads fail.

ADR-077 makes named @app.telemetry and @app.device entities publish retained "offline" to {prefix}/{device}/availability once a handler's retries are exhausted, and "online" on the next successful poll. Through 0.9.5 a sustained read failure only reached the {prefix}/status JSON blob, so a failing device kept a retained "online" and consumers presented its last reading as current.

Nothing to do if you want the new behaviour — it needs no code change:

# 0.9.6+: retries exhausted → "offline"; next successful poll → "online"
@app.telemetry("radon", interval=300, retry=2)
async def read_radon(ctx: cosalette.DeviceContext) -> dict[str, float]:
    return await ctx.adapter(SensorPort).read()

To restore the old behaviour on a noisy entity, disable it explicitly:

@app.telemetry("noisy", interval=60, unavailable_on=None)  # never marks offline
async def read_noisy(ctx: cosalette.DeviceContext) -> dict[str, float]: ...

To narrow it so a handler bug does not claim the device is unreachable — the recommended shape when you know your adapter's exception types:

# A KeyError from a malformed payload still reports on the error topic and in
# the status blob, but does not mark the device offline.
@app.telemetry("radon", interval=300, unavailable_on=(BleakError, TimeoutError))
async def read_radon(ctx: cosalette.DeviceContext) -> dict[str, float]:
    payload = await ctx.adapter(SensorPort).read()
    return {"radon": payload["radon_bq_m3"]}

Three things that do not change:

• @app.command is untouched. It keeps its opt-in None default — a command runs on demand, so a failed command says nothing about reachability.
• Root entities (name=None) are excluded from the automatic default and must pass an explicit unavailable_on to participate, because they publish to the flat {prefix}/availability and would otherwise declare the whole app unavailable.
• The status blob still carries why a device failed ("error", "circuit_open"). Availability stays a two-word vocabulary, so check {prefix}/status — not the availability topic — to tell a transport failure from a handler bug.

One related fix ships with it: a device marked unavailable is no longer republished as "online" by the post-reconnect reannounce, and it keeps its entry in the {prefix}/status roster instead of disappearing from it.

Fix: telemetry without retries now goes offline too (v0.11.0+)¶

Through 0.10.6 only retry exhaustion published "offline". A registration with the default retry=0, or one failing with an exception outside retry_on, has no retries to exhaust, so it stayed "online" however long it failed. It now goes "offline" on the first failed poll and back "online" on the next success.

If a single failed poll is too eager for an entity, give it retries (retry=2, retry_on=(...)) so only a sustained failure counts, or narrow or disable it with unavailable_on=.

See Transport Availability Signaling.

Fix: the circuit breaker now counts failures without retries (v0.11.0+)¶

Through 0.10.6 a circuit_breaker= counted only cycles whose retries were exhausted. With the default retry=0, or an exception outside retry_on, a failed poll never counted, so the breaker never opened, and a failed half-open probe left it probing every cycle. Every terminally failed cycle now counts once.

Behaviour change: a breaker configured with retry=0 now actually trips. After threshold consecutive failed polls the handler is skipped and the device reports "circuit_open" until a probe succeeds. Raise threshold, or remove circuit_breaker= if you relied on it never opening.

See Retry / Backoff.

Freshness: stale telemetry goes offline (v0.11.0+)¶

Behaviour change. ADR-080 adds a freshness watchdog. A named @app.telemetry entity with no fresh cycle for stale_after seconds now publishes retained "offline", even when no poll failed — a dead task, a hung poll without a timeout, or an error outside unavailable_on. The next fresh cycle publishes "online". When omitted, stale_after is derived as 2 × period + timeout × (retry + 1) + allowance × retry, where the allowance is the backoff's max_delay including jitter, at least 60 s; root entities are excluded.

The {prefix}/status heartbeat changes shape too: a stale entity reports "status": "stale" (outranking "error"), and every telemetry entry gains last_success_at and consecutive_failures. Consumers that compare the whole device entry, rather than reading status, need updating.

To opt out on an entity, disable it explicitly:

@app.telemetry("noisy", interval=60, stale_after=None)  # never marked stale
async def read_noisy(ctx: cosalette.DeviceContext) -> dict[str, float]: ...

A custom BackoffStrategy without a max_delay attribute is credited with 60 s per retry; set stale_after= explicitly if it sleeps longer. See Staleness.

Fix: offline availability survives an MQTT outage (v0.11.0+)¶

Through 0.10.6 a device that went "offline" while the broker was unreachable lost that publish. A reconnect skipped unavailable devices, and the first connect announced every registered device "online" (or, for most sources, published nothing), so the broker kept an older retained "online" for a device that was still failing. The first connect and every reconnect now publish each device's live availability: "offline" for unavailable devices, "online" for the rest. A ctx.mark_unavailable() made before the first connect is no longer cleared by the startup announce. No code change is needed; subscribers may see one extra, unchanged retained "offline" per unavailable device after a connect. See the 2026-10-02 amendment of ADR-012.

state_model= Return-Value Enforcement (v0.9.0+)¶

Breaking. ADR-068 makes the documented rule unconditional: if you declare state_model, published state is validated — on all four publishing archetypes. Through 0.8.x that only held for @app.device / @app.stream. Five things change.

1. Non-conforming payloads now fail at boot¶

A @app.telemetry / @app.command handler whose payload never matched its declared state_model raises ReturnValidationError on the first cycle after upgrade — usually a missing required field. The error is published to {prefix}/{name}/error and the state publish is suppressed, so the retained state topic goes stale instead of carrying a bad payload.

class Reading(BaseModel):
    sensor: str
    value: float


@app.telemetry("rx", interval=30, state_model=Reading)
async def rx():
    return {"sensor": "a"}  # 0.8.x: published verbatim. 0.9.0: raises.

Migration — one of two one-line choices:

• Fix the payload so it matches the model (add the missing field), or
• Drop state_model= to go back to unvalidated publishing.

2. state_model= outranks the return annotation¶

normalize_handler_return resolved get_return_annotation(func) or state_model; it now resolves state_model or get_return_annotation(func) (clause A). A handler declaring both, with different types, changes behaviour: the annotation used to govern, state_model= governs now.

# 0.8.x: dict[str, object] wins — TypeAdapter accepts anything, nothing is validated
# 0.9.0: Reading wins — the payload is validated
@app.telemetry("rx", interval=30, state_model=Reading)
async def rx() -> dict[str, object]: ...

3. The wire payload omits None instead of publishing null¶

Validated payloads dump with exclude_none=True on every archetype (clauses C and D), so an optional field the handler left out is an absent key, not an explicit null. This changes the @app.device / @app.stream wire payload for any state_model with optional fields:

// 0.8.x
{"sensor": "a", "brightness": null}
// 0.9.0
{"sensor": "a"}

Migration: update any consumer that reads those keys — Home Assistant value_templates in particular (a template that assumed the key is always present now needs value_json.get('brightness') or a default), plus retained-topic snapshots and exact-payload contract tests. It is also no longer possible to publish a deliberate null through a state_model.

4. The registration warning is an error under filterwarnings = ["error"]¶

Clause F emits a UserWarning at registration when state_model=M and the return annotation name different types. Under pytest's filterwarnings = ["error"] — a common strict-test configuration, and the one this repository uses — that warning is an error, so affected tests fail on upgrade. It broke 75 tests in cosalette's own suite, every one a genuine state_model=M + -> dict[str, object] contradiction.

Migration: remove the loose return annotation and leave state_model= as the sole contract.

# Before — warns (and fails under filterwarnings = ["error"])
@app.telemetry("rx", interval=30, state_model=Reading)
async def rx() -> dict[str, object]: ...


# After
@app.telemetry("rx", interval=30, state_model=Reading)
async def rx(): ...

5. -> None and -> M | None stay silent¶

These are not contradictions and are unaffected. -> M | None is the same contract with a suppress-publish case; -> None promises no return value at all, so clause A never overrides anything — state_model= there is channel metadata, and it is what gives an @app.command its AsyncAPI state channel. Keep it.

payload_model / state_model vs Type Annotations¶

Both forms are supported — explicit decorator metadata wins over annotation inference.

Explicit Decorator Metadata (v0.1.0+)¶

@app.command(
    "valve",
    payload_model=ValveCommand,  # inbound /set channel
    state_model=ValveState,  # outbound /state channel
)
async def handle_valve(payload: str) -> None:
    # Handler uses raw strings; schema enforced externally
    ...

Type Annotation Inference (v0.4.0+)¶

@app.command("valve")
async def handle_valve(
    cmd: Annotated[ValveCommand, Payload()],
) -> ValveState:
    # payload_model inferred from `cmd` parameter annotation
    # state_model inferred from return annotation
    ...

Schema inference priority:

• Commands (inbound /set): payload_model → injection plan (Annotated[T, Payload()] or payload: T) → {"type": "object"}
• Commands (outbound /state): state_model → return annotation → omitted (no noise for voids)
• Telemetry/devices: state_model → return annotation → {"type": "object"}

@app.device now accepts an explicit state_model= kwarg (previously only return-annotation inference was available for devices). payload_model= is also accepted on devices but is introspection-only — no device /set channel is emitted, so it does not affect schema output.

Prefer annotation inference for new code — it's more concise and the schema stays co-located with the handler signature. Do not declare both with different types: since 0.9.0 state_model= also outranks the return annotation at runtime, and the combination warns at registration. See state_model= Return-Value Enforcement.

@app.device Async Generator Requirement (v0.4.0+)¶

Before — plain async function (v0.1.0–v0.3.x):

@app.device("valve")
async def valve(ctx: cosalette.DeviceContext) -> None:
    @ctx.on_command
    async def handle(topic: str, payload: str) -> None:
        await ctx.publish_state({"state": payload})

    await ctx.publish_state({"state": "closed"})
    while not ctx.shutdown_requested:
        await ctx.sleep(30)

After — async generator with yield (v0.4.0+):

@app.device("valve")
async def valve(ctx: cosalette.DeviceContext):
    @ctx.on_command
    async def handle(topic: str, payload: str) -> None:
        await ctx.publish_state({"state": payload})

    await ctx.publish_state({"state": "closed"})
    yield  # Reaction boundary — reactors fire here
    while not ctx.shutdown_requested:
        await ctx.sleep(30)
        yield  # Reaction boundary

Why: yield creates reaction boundaries for domain-event reactors (@app.react). Reactors fire at execution boundaries before the next ctx.sleep().

See the Shared State guide for domain-event reactor patterns.

AsyncAPI and Manifest Introspection¶

app.asyncapi() (v0.2.0+)¶

Canonical AsyncAPI document generation — replaces older registry introspection wording.

import json
from pathlib import Path

# Generate AsyncAPI 3.0.0 document
doc = app.asyncapi()
Path("asyncapi.yaml").write_text(json.dumps(doc, indent=2))

Used by:

• cosalette schema dump CLI subcommand
• cosalette_manifest MCP tool
• CI/CD contract enforcement

Document structure:

• channels: MQTT topic definitions with payload schemas
• operations: Send/receive operations for each channel
• components.schemas: Pydantic model schemas as JSON Schema
• info.x-cosalette-contract-version: Contract-shape version (independent from app version)
• info.x-cosalette-topic-prefix: MQTT prefix the addresses were composed from (v0.9.3+, only when it differs from the app name)

Generated Addresses Honour mqtt.topic_prefix (v0.9.3+)¶

The v0.9.3 prefix change only affects apps that set mqtt.topic_prefix. Through v0.10.x, leaving it unset kept generated documents byte-identical to the pre-v0.9.3 output. From v0.11.0, every app's document changes for contract version "2"; see Availability Channels.

Before v0.9.3, generated AsyncAPI composed channel.address from App(name=...) and ignored the setting, so every derived artefact described topics the app never used. After upgrading:

# Regenerate anything you committed from a prefixed app
cosalette schema dump --app myapp.main:app --resolve-settings --config-file app.toml > asyncapi.yaml
cosalette schema acl --schema asyncapi.yaml > acl.conf
cosalette schema ha-discovery --schema asyncapi.yaml

Two behaviour changes to expect:

• Broker ACLs change. Generated ACLs previously granted topics under the app name while the app published under the prefix — they failed closed. Regenerate and redeploy them, or the app keeps being denied.
• Network-level enforcement starts working again. It previously filtered an app's slice by the address prefix rather than the app name, matching zero channels — so enforcement = "strict" silently stopped raising. It now filters by x-cosalette-app as intended, which means genuine contract violations that were being skipped will surface at startup. Run cosalette schema check before deploying.

Home Assistant object_id and unique_id are unchanged, including for multi-segment prefixes, so existing entities are not orphaned. See ADR-072.

Availability Channels in Generated AsyncAPI (v0.11.0+)¶

Affects every app. Generated documents now have one framework availability channel per device, telemetry, command and named stream, a matching publish…Availability operation, and x-cosalette-contract-version: "2". Unprefixed documents are no longer byte-identical to earlier releases.

The topics themselves do not change: the framework already published them. The document now describes them. After upgrading:

• Regenerate committed schema artefacts. Re-run cosalette schema dump (or schema init) for every schema baseline you commit, and update any golden-file test that compares generated output byte for byte.
• The retained {prefix}/_meta/registry snapshot changes. It is built from the same document, so it now lists the availability channels too. Tools that diff or cache that snapshot see the change once.
• Tools that iterate channels. A channel with x-cosalette-framework: "availability" has no x-cosalette-archetype. Skip it if your tooling expects every channel to be an app registration. Older cosalette loaders ignore the unknown key.

Home Assistant discovery, openHAB output and generated broker ACLs are unchanged, so nothing has to be redeployed to the broker or to Home Assistant. See ADR-086.

Schema Inference Priority¶

See payload_model / state_model vs Type Annotations above.

Exporting for External Tools¶

# Generate AsyncAPI YAML for schema enforcement
cosalette schema dump > asyncapi.yaml

# Validate with AsyncAPI CLI
asyncapi validate asyncapi.yaml

See Schema Enforcement for contract-first development workflows.

Testing Harness Updates¶

AppHarness.create() (v0.2.0+)¶

Before — manual app and double wiring:

from cosalette.testing import MockMqttClient, FakeClock

app = cosalette.App(name="testapp", version="1.0.0")
mqtt = MockMqttClient()
clock = FakeClock(0.0)
# ... manual wiring ...

After — AppHarness.create() with test doubles:

from cosalette.testing import AppHarness

harness = AppHarness.create(name="testapp")
# harness.app, harness.mqtt, harness.clock, harness.settings pre-wired

Testing Patterns (v0.2.0+)¶

Example:

import asyncio

import pytest
from cosalette.testing import AppHarness


@pytest.mark.asyncio
async def test_telemetry_publishes_on_interval():
    """Telemetry handler publishes state after interval elapses."""
    harness = AppHarness.create(name="testapp")

    @harness.app.telemetry("sensor", interval=30)
    async def sensor() -> dict[str, object]:
        return {"value": 42}

    # Orchestrate time advancement and shutdown
    async def advance_and_shutdown():
        await harness.advance_time(30)
        harness.trigger_shutdown()

    asyncio.create_task(advance_and_shutdown())
    await harness.run()

    # Assert published message
    messages = harness.mqtt.get_messages_for("testapp/sensor/state")
    assert len(messages) >= 1
    assert '"value": 42' in messages[0][0]

Fixture Conventions (v0.2.0+)¶

Register the pytest plugin in conftest.py:

pytest_plugins = ["cosalette.testing._plugin"]

This registers three fixtures:

See Testing for three-layer test patterns and shared fixture conventions.

Migration Checklist¶

Before upgrading cosalette:

1. Review the CHANGELOG for your target version
2. Run tests — task test:unit and task test:integration
3. Update handler signatures — add yield to @app.device handlers (v0.4.0+)
4. Migrate to typed payloads (optional but recommended) — replace raw strings with Pydantic models
5. Adopt AppHarness.create() in tests (v0.2.0+)
6. Audit every state_model= (v0.9.0+) — confirm the payload really matches the model, and drop return annotations that disagree with it
7. Regenerate AsyncAPI contracts — cosalette schema dump > asyncapi.yaml (required for v0.11.0+: contract version "2" adds availability channels)
8. Run quality gates — task check (lint + typecheck + tests)
9. Update downstream consumers — if MQTT topics changed due to Router prefixes, or if a state_model with optional fields now omits keys it used to publish as null (v0.9.0+)

Getting Help¶

• AI agent support: cosalette ai help <topic> — topics: contracts, router, testing, architecture
• GitHub Discussions: ff-fab/cosalette/discussions
• Issues: ff-fab/cosalette/issues