Transport Availability Signaling¶

When an adapter wraps a fallible transport — SSH, BLE, serial, HTTP — the device availability topic should reflect whether the transport is currently reachable. cosalette provides first-class support for this via unavailable_on on @app.command, @app.telemetry and @app.device, and ctx.mark_unavailable().

The Problem Without Framework Support¶

Without framework support, each adapter catches transport exceptions and manually constructs an "unavailable" state payload:

@app.command("display")
async def handle_display(ctx: DeviceContext) -> dict[str, object]:
    try:
        result = await ssh.read()
        return {"available": True, "brightness": result.brightness}
    except SSHError:
        return {"available": False, "brightness": None}  # manual unavailable payload

This pattern is repeated across every adapter with a fallible transport, has no standard MQTT availability topic, and doesn't integrate with Home Assistant's entity availability mechanism.

Static Form — unavailable_on¶

Declare which exception types represent transport failures directly on the decorator:

import cosalette

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


class SSHError(Exception):
    """Raised when the SSH connection to the device fails."""


@app.command("display", unavailable_on=(SSHError, TimeoutError))  # (1)!
async def handle_display(ctx: cosalette.DeviceContext) -> dict[str, object]:
    result = await ssh.read()  # (2)!
    return {"brightness": result.brightness}  # (3)!


app.run()

1. unavailable_on declares the exception tuple. Any exception in this tuple that escapes the handler is suppressed — it does not propagate.
2. If SSHError or TimeoutError is raised here, the framework intercepts it.
3. On success: the framework publishes the returned dict as device state.

What the framework does when a matching exception is raised:

1. Suppresses the exception (does not re-raise).
2. Publishes "offline" to wallpanel/display/availability (retained, QoS 1).
3. Logs a structured error payload to wallpanel/display/error.
4. Sets an internal _is_unavailable flag on the device context.

Dynamic Form — ctx.mark_unavailable()¶

For conditional unavailability — where you check reachability before attempting the operation — call ctx.mark_unavailable() directly from the handler body:

@app.command("sensor")
async def handle_sensor(ctx: cosalette.DeviceContext) -> dict[str, object]:
    if not await client.is_reachable():  # (1)!
        await ctx.mark_unavailable()  # (2)!
        return {}

    data = await client.read()
    return {"value": data.value}  # (3)!

1. Pre-flight reachability check — no exception needed.
2. mark_unavailable() publishes "offline" to the availability topic.
3. Next successful invocation triggers auto-recovery — for every archetype; see Defaults and Recovery by Archetype.

Dynamic Form — ctx.mark_available()¶

ctx.mark_available() is the symmetric counterpart to ctx.mark_unavailable(): it publishes "online" (retained, QoS 1) to the same availability topic and clears the internal _is_unavailable flag.

@app.telemetry("sensor", interval=30)
async def read_sensor(ctx: cosalette.DeviceContext) -> dict[str, object]:
    if not await client.is_reachable():
        await ctx.mark_unavailable()  # (1)!
        return {}

    if ctx._is_unavailable:
        await ctx.mark_available()  # (2)!

    data = await client.read()
    return {"value": data.value}

1. Publishes "offline", same as the command form.
2. Explicitly signals a recovery your handler detects itself. Declarative unavailable_on= failures on telemetry and device handlers otherwise recover automatically at their next successful work boundary.

mark_available() is a no-op when no HealthReporter is injected (e.g. in unit tests that construct a bare DeviceContext), mirroring mark_unavailable().

Auto-Recovery¶

@app.command handlers share an auto-recovery mechanism between the static and dynamic forms. After any successful command handler invocation — where no unavailable_on exception was raised and no early return without a matching exception — the framework:

1. Checks whether the internal _is_unavailable flag is set.
2. If yes: publishes "online" to the availability topic.
3. Resets the flag to False.

No explicit "come back online" call is needed in the handler.

MQTT events for two consecutive calls:

  Call 1: SSHError raised
    → wallpanel/display/availability  "offline"  (retained)
    → wallpanel/display/error         {...}

  Call 2: succeeds
    → wallpanel/display/state         {"brightness": 80}
    → wallpanel/display/availability  "online"   (retained)

Defaults and Recovery by Archetype¶

A successful poll is a genuine recovery signal for telemetry, which is why these archetypes now auto-recover (ADR-077, narrowing ADR-047's command-only scoping). ctx.mark_available() remains available for recovery your handler detects itself.

@app.telemetry("sensor", interval=30, retry=2)
async def read_sensor(ctx: cosalette.DeviceContext) -> dict[str, object]:
    return {"value": (await client.read()).value}

Retries exhausted publishes "offline"; the next successful poll publishes "online". Without retry= — or for an exception outside retry_on — there is nothing to exhaust, so the first failed poll publishes "offline". Narrow it when a handler bug should not claim the device is unreachable:

@app.telemetry("sensor", interval=30, unavailable_on=(BleakError, TimeoutError))
async def read_sensor(ctx: cosalette.DeviceContext) -> dict[str, object]:
    payload = await client.read()
    return {"value": payload["v"]}  # a KeyError here does NOT mark it offline

Freshness — stale_after¶

Failure-driven availability only fires when a poll fails. It cannot see a telemetry task that has died, a poll that hangs without a timeout, an init= that keeps failing, or an error that unavailable_on deliberately ignores. The reading then sits in Home Assistant as if it were current.

The freshness watchdog closes that gap (ADR-080). A named telemetry entity with no fresh cycle for its stale_after bound publishes retained "offline", and the next fresh cycle publishes "online" again. The bound is derived from the schedule unless you set it; streams opt in with @app.stream(..., stale_after=...).

@app.telemetry("radon", interval=300, stale_after=1800)  # offline after 30 min
async def read_radon(ctx: cosalette.DeviceContext) -> dict[str, float]: ...

The freshness mark is a separate availability source: clearing it never brings an entity online while a failure mark (unavailable_on, ctx.mark_unavailable()) still holds it offline, and vice versa.

Staleness describes which entities can go stale, how the bound is derived, the heartbeat and health file signals, and the exit_after_stale and restart_on_stale responses.

Scope — Device-Level¶

Availability state is device-scoped: all handlers that share the same device name share one availability state. If a device has multiple commands (e.g. via sub=), a single failure on any one of them marks the whole device offline.

Which Form to Use¶

Using Both Together¶

@app.command("display", unavailable_on=(SSHError,))
async def handle_display(ctx: cosalette.DeviceContext) -> dict[str, object]:
    if not await ssh.ping():
        await ctx.mark_unavailable()  # proactive check
        return {}
    return {"brightness": await ssh.read_brightness()}  # SSHError auto-handled

Home Assistant Integration¶

The {app}/{device}/availability topic is the standard MQTT availability topic expected by Home Assistant's MQTT integration. Configure it in your HA device configuration:

mqtt:
  sensor:
    - name: "Display Brightness"
      state_topic: "wallpanel/display/state"
      value_template: "{{ value_json.brightness }}"
      availability_topic: "wallpanel/display/availability"
      payload_available: "online"
      payload_not_available: "offline"

When the transport is unreachable, HA marks the entity as Unavailable instead of showing a stale value.

Relationship to HealthCheckRunner¶

HealthCheckRunner monitors adapter health on a polling schedule — it calls health_check() at a configurable interval and flips availability if the probe fails.

Transport availability signaling fires per command invocation — it reacts to real transport errors as they occur.

Both publish to the same {app}/{device}/availability topic and are fully complementary:

Testing¶

Use AppHarness to assert availability topic messages in integration tests:

import asyncio
import pytest
from cosalette import DeviceContext
from cosalette.testing import AppHarness


class TransportError(Exception):
    pass


@pytest.mark.asyncio
async def test_device_goes_offline_on_transport_error():
    harness = AppHarness.create(name="myapp")
    handler_called = asyncio.Event()

    @harness.app.command("sensor", unavailable_on=(TransportError,))
    async def handle(ctx: DeviceContext) -> None:
        handler_called.set()
        raise TransportError("unreachable")

    async def simulate() -> None:
        await asyncio.sleep(0.05)
        await harness.inject_command("sensor", "")
        await handler_called.wait()
        await asyncio.sleep(0.05)
        harness.trigger_shutdown()

    asyncio.create_task(simulate())
    await asyncio.wait_for(harness.run(), timeout=5.0)

    msgs = harness.messages_for("myapp/sensor/availability")
    assert "offline" in [m[0] for m in msgs]

See Testing for the full testing guide.