Error Handling¶

Cosalette treats errors as observable events — structured, published to MQTT, and designed for unattended daemons where operators cannot watch a terminal.

Design Principles¶

1. Structured payloads — every error is a JSON object with machine-readable fields
2. Fire-and-forget — publication failures never crash the daemon
3. Dual output — errors are both logged (WARNING) and published to MQTT
4. Error isolation — one device crashing does not affect others, and a dead task is reported and restarted by the task supervisor
5. Not retained — errors are events, not state

ErrorPayload¶

Every error is represented as a frozen dataclass before serialisation:

@dataclass(frozen=True, slots=True)
class ErrorPayload:
    error_type: str  # (1)!
    message: str  # (2)!
    device: str | None  # (3)!
    timestamp: str  # (4)!
    details: dict[str, object]  # (5)!

1. Machine-readable type string (e.g. "invalid_command", "timeout", "error").
2. Exception class name by default; str(exception) only when the type is disclosed (see Message Disclosure).
3. Device name when the error is device-scoped, None for global errors.
4. Wall-clock ISO 8601 timestamp: "2026-02-14T12:34:56+00:00".
5. Optional dict of additional context (default: empty).

Example Payload¶

{
    "error_type": "invalid_command",
    "message": "InvalidCommandError",
    "device": "blind",
    "timestamp": "2026-02-14T12:34:56+00:00",
    "details": {"raw_payload": "150"}
}

Topic Layout¶

Errors are published to two topics simultaneously:

graph LR
    E[Exception] --> B["build_error_payload()"]
    B --> G["Global: {prefix}/error"]
    B --> D["Device: {prefix}/{device}/error"]

The global topic receives every error, making it the single subscription point for fleet-wide monitoring (+/error). The per-device topic enables targeted monitoring and Home Assistant integration.

Building Error Payloads¶

The build_error_payload() function converts an exception into a structured payload:

from cosalette._errors import build_error_payload

error_type_map: dict[type[Exception], str] = {
    InvalidCommandError: "invalid_command",
    TimeoutError: "timeout",
    ConnectionError: "connection_lost",
}

payload = build_error_payload(
    error,
    error_type_map=error_type_map,
    device="blind",
    details={"raw_payload": raw},
)

Error Type Mapping¶

The error_type_map is a dict[type[Exception], str] that maps exception classes to machine-readable type strings:

• Exact match only — subclasses are not matched (this is intentional; it forces explicit registration of each error type)
• Fallback — unmapped exceptions receive the generic type "error"
• Pluggable — each application provides its own mapping

# Framework looks up the exact class
error_type = error_type_map.get(type(error), "error")

Framework Error Types¶

The framework labels the errors it raises and sanitises itself, so consumers can tell rejected input apart from a crash in handler code without any app configuration:

ReturnValidationError stays unlabelled (error): a handler that returns the wrong shape is an application fault.

Typed telemetry trigger parameters using Annotated[T, Payload()] use the same validation exception, so their failures also change to invalid_payload in 0.12.0. Their message follows the same disclosure rules as typed command validation: with PayloadValidationError listed in disclose_messages_for it contains a sanitized summary, otherwise the class name.

The app's error_type_map is merged with these entries and the framework entries win. When an app maps one of these types to a different label, the app logs one WARNING at startup and the framework label is used:

error_type_map entry PayloadValidationError -> 'validation' is overridden by the framework label 'invalid_payload'

Drop the entry from your map to silence it. An entry with the same label as the framework is harmless and logs nothing. See ADR-011.

Log Severity for Rejected Input¶

Command input the framework rejects is expected bad input, not a fault. It is logged once, at WARNING, without a traceback:

WARNING cosalette._errors: Publishing error [id=825f98e2ac5f type=invalid_payload device=display]: Payload validation failed (parameter 'cmd'): [field]: type=literal_error

This covers a typed @app.command payload that fails JSON decoding or schema validation before the handler runs, and the sub-command errors invalid_json, missing_sub_key and unknown_sub_command. No ERROR line is written, so alert rules on level=ERROR do not fire on a typo. For typed command input the full chained traceback, including the pydantic detail, is logged at DEBUG by cosalette._runners._command_runner (see Per-logger Levels). An unknown sub-command also keeps its local WARNING with the raw value, since the published message only carries a fingerprint.

A typed telemetry trigger payload (Annotated[T, Payload()] on a triggerable=True handler) that fails the same way is handled alike: one WARNING without traceback, invalid_payload on the error topic, and the traceback at DEBUG from cosalette._runners._telemetry_runner. The handler does not run for that trigger. A rejected trigger is not a failed cycle: it does not count toward the failure streak or consecutive_failures and does not set the device status to "error" (ADR-082). The next scheduled run or valid trigger runs normally.

Exceptions raised by handler code are unchanged: an ERROR line plus the WARNING with the traceback. That includes a PayloadValidationError the handler or an application dependency raises itself, for example while validating a nested payload. For telemetry, such an exception is a failed cycle as before, and so is a typed trigger parameter that fails to bind on a scheduled run (a non-optional T bound to None); in a coalescing group only that member's cycle fails. Unknown sub-command warnings include only the value's type, length and non-reversible fingerprint, so raw values stay out of local logs. The error topic payload is unchanged; only the local log and health differ.

Message Disclosure¶

error_type labeling and whether the raw str(error) is published are two separate decisions. Because error topics are broker-visible, an unredacted exception message can leak secrets — credentials embedded in a URL, a filesystem path, a hostname. Nothing is disclosed by default: every error payload, the framework's own included, carries its error_type label and the exception class name as message. build_error_payload() discloses a message in exactly two cases:

1. Verbose publishing — discloses every exception's message. It takes effect only when the app allows it with App(allow_error_publish_verbose=True) and the operator sets MqttSettings.error_publish_verbose (MQTT__ERROR_PUBLISH_VERBOSE=true). When the operator sets it on an app that does not allow it, the setting is ignored and one WARNING is logged at startup.
2. disclose_messages_for (a frozenset[type[Exception]] | None) — a type's message is published only if that exact type is a member of the set; subclasses are not matched. None (the default) behaves like frozenset(): nothing is disclosed.

error_type_map only assigns labels. Mapping a type never publishes its message.

# Default: label + class name for every error.
app = cosalette.App(
    name="caldates2mqtt",
    error_type_map={CalDavConnectionError: "caldav_connection_error"},
)

# Keep the framework's own sanitised messages, disclose one app type.
app = cosalette.App(
    name="caldates2mqtt",
    error_type_map={CalDavConnectionError: "caldav_connection_error"},
    disclose_messages_for=cosalette.FRAMEWORK_DISCLOSED_ERRORS
    | {CalDavConnectionError},
)

cosalette.FRAMEWORK_DISCLOSED_ERRORS (InvalidJsonError, MissingSubKeyError, UnknownSubCommandError, PayloadValidationError) lists the messages the framework builds and sanitises itself, including typed telemetry trigger validation. The builtin TimeoutError is deliberately not in it: library code raises it too, with text the framework does not control. Command watchdog timeouts carry the framework message command handler exceeded N.Ns watchdog; a TimeoutError raised inside a handler keeps its own text. Either is published only when the app lists TimeoutError itself.

Undisclosed exception types publish the class name only — the full message and traceback are still logged locally under a correlation id. See ADR-091 for the rationale.

Redacting Disclosed Messages¶

Disclosure is all or nothing per type: once a type is disclosed, its whole message is published. When a disclosed message can still carry a token, a MAC address or a URL with credentials, pass App(redact=...). It is applied after the disclosure decision, verbose publishing included, so it never widens what is published:

import re

app = cosalette.App(
    name="airthings2mqtt",
    disclose_messages_for=frozenset({CloudApiError}),
    redact=[r"token=[^&\s]+", re.compile(r"([0-9A-F]{2}:){5}[0-9A-F]{2}", re.I)],
)
# CloudApiError("GET /v1?token=abc failed") → message "GET /v1?[REDACTED] failed"

# or any str -> str callable
app = cosalette.App(
    name="velux2mqtt", redact=lambda text: text.replace(KEY, "[REDACTED]")
)

• redact is None (default, off), a callable str -> str, or an iterable of regular expressions (str or compiled re.Pattern); every match becomes [REDACTED].
• Patterns are compiled when the App is created: an invalid pattern raises ValueError, a wrong type (including a single bare str) raises TypeError.
• The same redactor runs on every record the log handlers cosalette installs write — see Logging.
• A redactor that raises lets the text through unchanged and logs one WARNING for the whole process. Redaction is a safety net on top of the disclosure policy, not a replacement for it.

See ADR-085.

ErrorPublisher Service¶

The ErrorPublisher wraps build_error_payload() with fire-and-forget MQTT publication:

@dataclass
class ErrorPublisher:
    mqtt: MqttPort
    topic_prefix: str
    error_type_map: dict[type[Exception], str] = field(default_factory=dict)
    clock: Callable[[], datetime] | None = field(default=None)
    verbose: bool = False
    disclose_messages_for: frozenset[type[Exception]] | None = None
    redact: Callable[[str], str] | None = None  # ADR-085, from App(redact=...)

    async def publish(self, error: Exception, *, device: str | None = None) -> None: ...

The entire pipeline — build → serialise → publish — is wrapped in fire-and-forget semantics:

async def publish(self, error, *, device=None):
    try:
        payload = build_error_payload(error, ...)
        payload_json = payload.to_json()
    except Exception:
        logger.exception("Failed to build error payload")
        return  # (1)!

    logger.warning("Publishing error: %s", payload.message)
    await self._safe_publish(global_topic, payload_json)
    if device is not None:
        await self._safe_publish(device_topic, payload_json)

1. Even if payload construction fails (unexpected), the daemon continues. This is the "fire-and-forget" guarantee — error reporting must never become a source of errors itself.

Fire-and-Forget Semantics¶

Every MQTT publication in the error pipeline is wrapped in a _safe_publish method:

async def _safe_publish(self, topic: str, payload: str) -> None:
    try:
        await self.mqtt.publish(topic, payload, retain=False, qos=1)
    except Exception:
        logger.exception("Failed to publish error to %s", topic)

This means:

• A broker outage does not crash device tasks
• A serialisation bug does not propagate to callers
• The worst case is a lost error event (logged locally)

Task Supervision¶

Every task the framework starts — @app.device generators, telemetry entities and coalescing groups, periodic handlers and stream handlers — runs under a task supervisor (ADR-081). One device crashing does not take the others down, and a dead task is never silent.

A task counts as failed when it ends with an exception, or when something outside the framework cancels it. A normal return is not a failure, and neither is the cancellation the framework itself issues during shutdown or an adapter restart.

When a task fails, the supervisor:

1. Logs one CRITICAL line with the traceback, from the logger cosalette._supervisor: Task 'device:blind' died (entities: blind): Connection refused.
2. Publishes one error payload (see below).
3. Marks the task's entities offline with the availability source "supervisor", and sets their heartbeat status to "error". Root streams report heartbeat status without an availability topic: see Stream failures.
4. Applies the app's on_task_failure policy.

The on_task_failure policy¶

app = cosalette.App(
    name="velux2mqtt",
    version="1.0.0",
    on_task_failure="restart",  # (1)!
    task_max_restarts=3,  # (2)!
    task_restart_window=300.0,  # (3)!
)

1. "restart" (default), "exit" or "ignore".
2. Restarts allowed per registration (per group for a coalescing group, per member for a group member's init=). 0 makes "restart" exit on the first failure.
3. Seconds without a failure after which the restart count resets to 0.

A shutdown triggered by the supervisor raises cosalette.TaskSupervisionError from App.run() (attributes task_name, restart_count, internal; the original exception is its __cause__). The CLI maps it to EXIT_TASK_FAILURE (4), so a container orchestrator with a restart policy restarts the whole process.

A re-created telemetry task or coalescing group waits one full interval before its first poll, so a crash on start-up cannot turn into a tight poll loop. When a re-created task recovers — a device reaches its first yield, a telemetry task completes a cycle — the supervisor clears its mark and the entity is online again.

Coalescing-group member init= failures¶

A coalescing group runs as one task, but a member whose init= raises does not fail it. One flaky sensor out of range at start-up must not take its healthy group mates offline, nor crash-loop the whole app. The failure is isolated to the member:

• the other members keep polling on their schedule;
• the supervisor logs one CRITICAL line, Init of group member 'radon' (task 'group:airthings') failed: ..., publishes one task-failure payload whose details carry "task": "group:airthings", "member": "radon" and "phase": "init", and marks only that member offline (source "supervisor", status "error");
• the member's triggers wait until it joins the schedule.

The app's on_task_failure policy applies to the member on its own:

If every member of a group ends up offline with no retry pending, the group logs one ERROR line and idles until shutdown; it neither exits nor restarts. A member's budget survives a re-created group — whether the supervisor or an adapter restart re-created it — so re-creation never resets it.

Any other failure of a group task, such as an exception raised outside a member's init=, still fails the whole group as described above. An ungrouped telemetry entity whose init= raises fails its own task as usual.

Stream failures¶

A stream is not in Home Assistant discovery, but a named stream owns a retained {prefix}/{stream}/availability topic. A crashed stream is reported like any task:

• the same CRITICAL line and the same task-failure payload as any task;
• "offline" on {prefix}/{stream}/availability under the supervisor source, and status "error" in the {prefix}/status heartbeat;
• the first item from the re-created stream clears the supervisor mark; it returns to "online" and "ok" when no other source holds it offline;
• the same on_task_failure policy, restart budget and exit code 4.

A root stream (@app.stream() without a name) is heartbeat-only: its payload goes to {prefix}/error only and it never publishes {prefix}/availability. If entities depend on a stream's data, declare @app.stream("feed", feeds=["radon"]) so they go offline with the stream; see Streaming and ADR-081 (2026-10-03 amendment).

The task-failure error payload¶

The supervisor publishes the payload through the regular ErrorPublisher, so error_type_map and the disclosure policy apply as usual. details identifies it as a task failure:

{
    "error_type": "error",
    "message": "ConnectionRefusedError",
    "device": "blind",
    "timestamp": "2026-02-14T12:34:57+00:00",
    "details": {"task_failure": true, "task": "device:blind"}
}

Every payload goes to the global {prefix}/error topic. A task that serves one entity also publishes to that entity's error topic. A coalescing group publishes to the global topic only, with details.entities listing its members; a periodic task, which has no entity, publishes to the global topic without a device. A root entity, such as a root stream registered without a name, publishes to the global topic only; its device field carries the handler name.

Per-Cycle Isolation in Telemetry¶

For telemetry devices, isolation is per polling cycle — a single failed reading does not stop the polling loop:

# Simplified TelemetryRunner.run_telemetry (see _telemetry_runner.py)
async def run_telemetry(self, reg, ctx, error_publisher, health_reporter):
    last_error_type = None
    while not ctx.shutdown_requested:
        try:
            result = await reg.func(ctx)
            await ctx.publish_state(result)
            if last_error_type is not None:
                logger.info("Telemetry '%s' recovered", reg.name)
                last_error_type = None
                health_reporter.set_device_status(reg.name, "ok")
        except asyncio.CancelledError:
            raise
        except Exception as exc:
            if type(exc) is not last_error_type:
                await error_publisher.publish(exc, device=reg.name)
            last_error_type = type(exc)
            health_reporter.set_device_status(reg.name, "error")
        await ctx.sleep(reg.interval)

The telemetry loop uses state-transition deduplication: only the first error of each type is published to MQTT. If the same exception type recurs on subsequent cycles, the publish is suppressed — preventing error floods from a persistently broken sensor. When the device recovers (a successful poll after a failure), recovery is logged at INFO level and the device health status is restored to "ok" in the heartbeat payload.

A suppressed error is not forgotten (ADR-082). While the same error persists, the 2^nd, 4^th, 8^th, ... failure within the first error_reminder_interval (default one hour), and then one failure per interval, is logged at WARNING and republished as a reminder. Every telemetry error payload carries the failure streak in details:

"details": {"count": 25, "first_seen": "2026-10-02T08:15:00+00:00"}

The recovery line states how long the outage lasted:

Telemetry 'radon' still failing after 2h00m00s (25 consecutive failures): ...
Telemetry 'radon' recovered after 2h10m00s (26 failed cycles)

Pass App(error_reminder_interval=None) to keep only the onset and the recovery line.

A trigger payload the framework rejects is operator input, not a failed cycle, and stays outside this flow (see Log Severity for Rejected Input).

When retry > 0 is configured on a telemetry handler, the framework wraps the handler call in a retry loop before reaching the error publication path shown above. Retry attempts are logged at WARNING level but are not published to the error topic — only the final failure (after all retries are exhausted) triggers the standard error-publish-and-deduplicate flow. The retry counter is cumulative across poll cycles and resets on success. An optional CircuitBreaker can short-circuit retries when a backend is persistently unavailable. See the Retry / Backoff guide and ADR-024 for details.

The retry machinery only engages when the handler raises. A handler that hangs mid-await — a BLE read, a serial .read(), an HTTP call with no internal timeout — never raises, so the retry and error-publication paths are never reached. The timeout= parameter converts a hang into a TimeoutError: because TimeoutError is a subclass of OSError (PEP 3151), it flows through the existing retry/backoff/error pipeline with no extra configuration. By default, every interval-based telemetry handler has an implicit backstop equal to its resolved interval; pass timeout=None to disable it for legitimately long-running handlers. See the Timeout backstop section and ADR-024 Decision 6 for details.

See Also¶

• MQTT Topics — topic layout and retained/non-retained rationale
• Health & Availability — complementary health reporting
• Device Archetypes — error isolation per device type
• Logging — errors are also logged at ERROR level
• ADR-011 — Error Handling and Publishing
• ADR-091 — Labelling-Only Error Type Map and Secure-by-Default Message Disclosure
• ADR-085 — Optional Redaction Hook for Logs and Disclosed Error Messages
• ADR-024 — Telemetry Retry/Backoff
• ADR-081 — Supervision of Framework-Started Tasks