Health & Availability¶

Cosalette provides three levels of health reporting — app-level heartbeats, per-device availability, and adapter health checks — backed by MQTT's Last Will and Testament (LWT) for crash detection.

Three Levels of Health¶

graph TB
    subgraph "App Level"
        A["LWT: broker publishes 'offline' on crash"]
        B["Heartbeat: JSON with uptime + device map"]
    end
    subgraph "Device Level"
        C["availability: 'online' / 'offline' per device"]
    end
    subgraph "Adapter Level"
        E["HealthCheckable: periodic readiness probe"]
    end
    A --> S["{prefix}/status"]
    B --> S
    C --> D["{prefix}/{device}/availability"]
    E -->|"failure toggles"| C

Last Will and Testament (LWT)¶

The LWT is an MQTT feature where the client tells the broker: "If I disconnect unexpectedly, publish this message on my behalf." Cosalette uses it to guarantee that a crash results in a visible "offline" status.

WillConfig¶

The build_will_config() function creates a WillConfig for the app's LWT:

from cosalette._health import build_will_config

will = build_will_config("velux2mqtt")
# → WillConfig(
#       topic="velux2mqtt/status",
#       payload="offline",
#       qos=1,
#       retain=True,
#   )

This WillConfig is passed to MqttClient during construction and translated to the broker-specific LWT format at connection time.

Crash Detection Flow¶

sequenceDiagram
    participant App
    participant Broker
    participant Subscriber as Monitoring Tool

    App->>Broker: CONNECT (will="offline" on velux2mqtt/status)
    App->>Broker: PUBLISH velux2mqtt/status = {"status": "online", ...}
    Note over App: App crashes / network lost
    Broker->>Subscriber: PUBLISH velux2mqtt/status = "offline" (LWT)

Structured Heartbeat¶

The app heartbeat is a JSON payload published to {prefix}/status:

{
    "status": "online",
    "uptime_s": 3600,
    "version": "0.3.0",
    "devices": {
        "blind": {"status": "ok"},
        "temperature": {
            "status": "error",
            "last_success_at": "2026-10-02T08:15:00+00:00",
            "consecutive_failures": 2,
            "last_error": "TimeoutError",
            "failing_since": "2026-10-02T08:20:00+00:00"
        }
    }
}

HeartbeatPayload Fields¶

Uptime Uses Monotonic Clock¶

Uptime is measured via ClockPort (backed by time.monotonic() in production):

uptime = int(self.clock.now() - self._start_time)

Two Payload Formats on One Topic¶

The {prefix}/status topic carries two different payload formats:

velux2mqtt/status → "offline"

{"status": "online", "uptime_s": 3600, "version": "0.3.0", "devices": {...}}

Consumers can distinguish them by attempting JSON parse. The LWT is always a plain string; the heartbeat is always valid JSON.

Per-Device Availability¶

Each device gets its own availability topic, published automatically by the HealthReporter:

A named stream owns {prefix}/{stream}/availability like a device: a task failure publishes "offline" with heartbeat status "error", and the first item after the restart clears the supervisor mark, publishing "online" when no other source holds it offline. Streams also support ctx.mark_unavailable() / ctx.mark_available(), an opt-in stale_after=, and feeds=[...] to hold the entities they supply offline with them. A root stream is heartbeat-only and never publishes {prefix}/availability. See Stream failures and Streaming.

# Published automatically by the framework
await health_reporter.publish_device_available("blind")
# → "online" to velux2mqtt/blind/availability (retained, QoS 1)

This is directly compatible with Home Assistant's MQTT availability configuration — no custom templates needed.

Sub-Entity Availability¶

Devices and streams can spawn temporary sub-components with their own availability lifecycle using ctx.sub_entity() (ADR-031). Entering the context manager publishes "online" to {prefix}/{device}/{sub}/availability; exiting publishes "offline" and clears retained state — mirroring the device-level pattern one topic level deeper.

A sub-entity of a root stream uses {prefix}/{sub}/availability, because a root stream's topic base is the app prefix.

Sub-entity availability is published by the context manager itself, not by the HealthReporter (ADR-031). It follows the async with block and nothing else:

• A crash takes it "offline", because the handler leaves the block. It comes back "online" only when the re-created handler enters the block again.
• It is not in the {prefix}/status heartbeat devices map. A reconnect does not re-announce it, and the app's graceful-shutdown sweep does not touch it.
• An adapter health-check failure, ctx.mark_unavailable(), stale_after= and feeds= hold the parent entity offline but leave the sub-entity's topic as it is.

For a device or named stream parent, a Home Assistant entity can combine the parent's availability topic with the sub-entity's topic. Set availability_mode: all so both must be online; the default latest follows whichever topic published last and can hide a parent's outage:

mqtt:
  sensor:
    - name: "Radio battery"
      state_topic: "myapp/radio/battery/state"
      value_template: "{{ value_json.level }}"
      availability:
        - topic: "myapp/radio/availability"
        - topic: "myapp/radio/battery/availability"
      availability_mode: all

A root stream has no parent availability topic: {prefix}/availability is app-wide and does not reflect its health. Use a named stream for the parent-topic approach; root stream health remains in the heartbeat.

See ADR-031 — Sub-Entity Context Manager for the full design.

Orphaned Retained-Topic Cleanup¶

When an entity is removed from configuration between restarts, its retained state and availability topics would otherwise linger on the broker forever — Home Assistant and other subscribers keep rendering a "ghost" device that no longer exists. Apps with a configured store= clear these orphaned topics automatically on the first MQTT connect.

The framework persists the resolved entity set under a reserved, prefix-namespaced store key. On the next run's first connect it diffs the current registrations against the previous run's set and publishes an empty (zero-byte) retained message — MQTT's standard "clear retained" convention, the same one ctx.sub_entity() uses on exit — to each removed entity's state and availability topics.

See ADR-048 — Clear Orphaned Retained Topics and ADR-049 — Default store path resolution for the full design and alternatives considered.

HealthReporter Service¶

The HealthReporter manages all health-related publications:

@dataclass
class HealthReporter:
    mqtt: MqttPort
    topic_prefix: str
    version: str
    clock: ClockPort

Key methods:

Fire-and-Forget Publishing¶

All health publications are wrapped in _safe_publish():

async def _safe_publish(self, topic, payload, *, retain=True):
    try:
        await self.mqtt.publish(topic, payload, retain=retain, qos=1)
    except Exception:
        logger.exception("Failed to publish health to %s", topic)

Health reporting must never crash the application. A broker outage means health data is temporarily lost, but service continues.

Graceful Shutdown Sequence¶

During Phase 4 teardown, the HealthReporter.shutdown() method publishes offline status for everything:

async def shutdown(self):
    for device in list(self._devices):
        await self._safe_publish(f"{prefix}/{device}/availability", "offline")
    await self._safe_publish(f"{prefix}/status", "offline")
    self._devices.clear()

This ensures that a clean shutdown results in the same "offline" state as a crash (via LWT). Subscribers see a consistent state regardless of how the application stopped.

Fleet Monitoring¶

Subscribe to wildcard topics to monitor multiple bridges:

# All app statuses across the fleet
mosquitto_sub -t '+/status' -v

# All device availability for one app
mosquitto_sub -t 'velux2mqtt/+/availability' -v

The retained nature of status and availability topics means new subscribers immediately receive the current state of every app and device.

Adapter Health Checks¶

The LWT and heartbeat mechanisms detect crashes — the app process dies and the broker publishes "offline". But what about adapters that are running but broken? A BLE adapter can enter a wedged state (BlueZ daemon crash, hardware reset) where connections fail indefinitely, yet the process stays alive.

Adapter health checks (ADR-028) address this gap. Adapters implement the HealthCheckable protocol, and the framework probes them periodically.

Implementing HealthCheckable¶

Add a single async method to any adapter:

class BleAdapter:
    """BLE adapter with health check support."""

    async def connect(self) -> None: ...
    async def read(self, mac: str) -> Reading: ...

    async def health_check(self) -> bool:
        """Return True if BLE stack is responsive."""
        scanner = BleakScanner()
        await scanner.discover(timeout=5)
        return True
        # Exceptions propagate to HealthCheckRunner, which treats them as failure

HealthCheckable is a @runtime_checkable Protocol — the framework detects it via isinstance() after adapter lifecycle entry. No registration or configuration needed.

How It Works¶

sequenceDiagram
    participant Runner as HealthCheckRunner
    participant Adapter
    participant Reporter as HealthReporter
    participant MQTT

    Note over Runner: Startup (before device tasks)
    Runner->>Adapter: health_check()
    Adapter-->>Runner: True
    Note over Runner: Adapter healthy, devices stay online

    Note over Runner: Periodic loop (every 30s)
    Runner->>Adapter: health_check()
    Adapter-->>Runner: False (or timeout/exception)
    Runner->>Reporter: publish_device_unavailable("sensor")
    Reporter->>MQTT: "offline" → sensor/availability

    Note over Runner: Next check
    Runner->>Adapter: health_check()
    Adapter-->>Runner: True
    Runner->>Reporter: publish_device_available("sensor")
    Reporter->>MQTT: "online" → sensor/availability

The lifecycle in detail:

1. Startup check — one probe per adapter before device tasks launch. Failed adapters start with their devices marked "offline", but device tasks still start (health checks are informational, not blocking).
2. Periodic loop — probes every health_check_interval seconds (default 30, configurable on App(), None to disable entirely).
3. Timeout — each probe has a timeout of interval / 2. A hanging health_check() is treated as failure without blocking other adapters.
4. Availability toggle — on failure, all devices that depend on the adapter are set to "offline"; on recovery, they return to "online" (unless the adapter's restart budget is spent — see Auto-Restart below).
5. Telemetry continues — health checks are informational. Telemetry polling continues even when the adapter is marked unhealthy.

Adapter-to-Device Mapping¶

The framework automatically maps adapters to their dependent devices using DI introspection. When adapter X fails a health check, only devices that inject X via their type annotations go offline — other devices are unaffected.

app = App("myapp", health_check_interval=30.0)
app.adapter(BlePort, BleAdapter)  # implements HealthCheckable


@app.telemetry("temperature", interval=60)
async def temperature(ctx: DeviceContext, ble: BlePort) -> dict[str, object]:
    # If BleAdapter.health_check() fails, "temperature" goes offline
    return await ble.read("AA:BB:CC:DD:EE:FF")


@app.telemetry("cpu_temp", interval=60)
async def cpu_temp(ctx: DeviceContext) -> dict[str, object]:
    # No BlePort dependency — unaffected by BLE health check failures
    return {"celsius": read_cpu_temp()}

Streams are mapped too. A Stream[T] handler depends on the StreamablePort[T] adapter behind it, plus any adapter it injects directly. A failing check takes a named stream offline (a root stream shows it in its heartbeat status), and an adapter restart cancels the stream and re-creates it, which closes the port and opens it again. A stream with stale_after= that goes stale requests the same restart under restart_on_stale (Staleness).

Failure Counting¶

Each adapter's health state is tracked via AdapterHealthStatus:

The consecutive_failures counter drives the auto-restart threshold — when it reaches restart_after_failures, a restart is triggered.

Log Deduplication¶

Health check logging follows the same deduplication pattern as error handling:

This prevents log flooding during sustained adapter outages while still providing clear visibility into failure onset and recovery.

Auto-Restart¶

When an adapter fails health checks repeatedly, the framework can automatically restart it — exit its async context manager, wait a cooldown, re-enter, and recreate device tasks. This handles transient hardware wedges (BLE daemon crash, serial port reset) without operator intervention. An adapter that only holds software state can offer a reset() method instead of a context manager.

Configuration¶

Auto-restart is controlled by four parameters on App():

app = App(
    "myapp",
    health_check_interval=30.0,
    restart_after_failures=5,  # restart after 5 consecutive failures
    max_restarts=3,  # give up after 3 restarts
    restart_cooldown=5.0,  # 5s between exit and re-enter
    sustained_health_reset=300.0,  # 5 min healthy resets counter
)

Restart Sequence¶

When consecutive failures reach the threshold:

sequenceDiagram
    participant Runner as HealthCheckRunner
    participant Wiring as _on_restart callback
    participant Adapter
    participant Devices as Device Tasks

    Runner->>Runner: consecutive_failures >= restart_after_failures
    Runner->>Wiring: on_restart_needed(adapter_type, adapter)
    Wiring->>Devices: cancel_tasks_for_adapter()
    Wiring->>Adapter: __aexit__() (best-effort)
    Note over Adapter: cooldown period (restart_cooldown)
    Wiring->>Adapter: __aenter__()
    Wiring->>Adapter: health_check() (verification)
    Wiring->>Devices: start_device_tasks_for_names()
    Wiring-->>Runner: True (success)
    Runner->>Runner: reset consecutive_failures, increment restart_count

A failed restart (re-entry or post-restart health check fails) still counts toward max_restarts, and the adapter's devices stay offline. The failure counter starts again from zero, so the next attempt follows another restart_after_failures failed checks. When the last attempt in the budget fails, the adapter is marked restart_exhausted and its devices stay offline until the process restarts.

An adapter can also pass a health check on its own between attempts. If its __aenter__ succeeded (only the post-restart health check failed), the devices the failed attempt cancelled are re-created and go back online. If __aenter__ failed, the adapter is outside its context: a passing check counts as failed and the next restart attempt re-enters it. A restart_exhausted adapter never goes back online, even when its health check passes again.

restart_cooldown is only the pause between __aexit__ and __aenter__. The failure threshold alone sets how far apart two restarts are.

Restart by reset()¶

An adapter without __aenter__/__aexit__ is restartable when it has a reset() method (ADR-084). The restart cancels the dependent device tasks, waits restart_cooldown, awaits reset(), runs the post-restart health check and re-creates the tasks. The framework never enters or exits such an adapter. A reset() that raises is a failed restart and counts toward max_restarts.

class AirthingsClient:
    async def health_check(self) -> bool: ...

    async def reset(self) -> None:
        """Drop the BLE client and parser; the next poll reconnects."""
        self._client = None

reset() should rebuild software state only, such as a client object, a parser or a cache. When an adapter has both a context manager and reset(), the restart uses the context manager and never calls reset().

Restart on Stale¶

A health check can pass while the adapter delivers no data. With App(restart_on_stale=True), a telemetry entity or a stream with stale_after= that goes stale also requests a restart of the restartable adapters it depends on. The request skips restart_after_failures, but it follows the restart sequence above and counts toward max_restarts. See Staleness for when it fires and how it combines with exit_after_stale.

Opting Out¶

By default, all adapters with HealthCheckable and either a lifecycle (__aenter__/__aexit__) or a reset() method are eligible for auto-restart. Set restartable = False on the adapter class to opt out:

class CriticalAdapter:
    """Adapter that must not be restarted mid-session."""

    restartable = False

    async def __aenter__(self) -> Self: ...
    async def __aexit__(self, *exc: object) -> None: ...
    async def health_check(self) -> bool: ...

The opt-out is logged once at INFO. A health-checkable adapter with neither __aenter__/__aexit__ nor reset() that does not opt out logs a WARNING at startup, since it can never be restarted; set restartable = False to acknowledge that and silence it. Either way the framework never restarts it: when it reaches restart_after_failures, the runner logs one WARNING for that unhealthy episode and its devices stay offline until a health check passes again.

Sustained Health Reset¶

If an adapter stays healthy for sustained_health_reset seconds (default 5 min), its restart_count resets to zero. This allows adapters that experience rare transient failures to get a fresh restart budget without accumulating toward max_restarts.

See Also¶

• MQTT Topics — complete topic map and retention rules
• Error Handling — structured error events (complementary to health)
• Staleness — what goes stale and how the app responds
• Lifecycle — when availability is published (Phases 2 and 4)
• Hexagonal Architecture — ClockPort for monotonic uptime
• ADR-012 — Health and Availability Reporting
• ADR-028 — Adapter Health Check Protocol
• ADR-029 — Adapter Auto-Restart Strategy
• ADR-084 — Adapter reset() Restart Protocol and Restart on Stale
• ADR-031 — Sub-Entity Context Manager