Handling Errors¶

This guide covers two topics: fixing framework errors when cosalette raises a registration or runtime exception, and customizing error classification using build_error_payload() with domain-specific exceptions.

For the complete error catalog (message text, location, cause), see Error Taxonomy.

Fixing Framework Errors¶

Decorator Parentheses¶

Always use parentheses on @app.device and @app.command, even with no arguments:

# Wrong
@app.device
async def my_device(ctx: DeviceContext) -> dict[str, object]: ...


# Correct
@app.device()
async def my_device(ctx: DeviceContext) -> dict[str, object]: ...

Async init Callback¶

The init= callback runs during synchronous bootstrap — it cannot be async:

# Wrong
async def setup_sensor():
    return SensorClient()


@app.device(init=setup_sensor)  # TypeError!
async def sensor(ctx: DeviceContext) -> dict[str, object]: ...


# Correct
def setup_sensor():
    return SensorClient()


@app.device(init=setup_sensor)
async def sensor(ctx: DeviceContext) -> dict[str, object]: ...

init Result Shadows Injectable¶

Wrap the return value in a domain-specific type instead of returning a framework type (AppContext, MqttPort, etc.) directly.

Bool Parameters (Type Guard)¶

Pass a numeric literal, not a boolean:

# Wrong
Pt1Filter(tau=True, dt=0.1)  # TypeError

# Correct
Pt1Filter(tau=1.0, dt=0.1)

Similarly, MedianFilter(window=5) — pass an integer, not True/False.

Handler Annotations¶

Every handler parameter must have a concrete type annotation:

# Wrong — missing annotation
@app.device()
async def sensor(ctx):  # TypeError!
    ...


# Wrong — *args
@app.device()
async def sensor(*args: DeviceContext) -> dict[str, object]:  # TypeError!
    ...


# Correct
@app.device()
async def sensor(ctx: DeviceContext) -> dict[str, object]: ...

Depends() — Async Dependency¶

Dependencies must be synchronous — the framework cannot await them while building handler kwargs:

# Wrong — async dependency
async def get_client() -> Client: ...


# Wrong — async __call__
class GetClient:
    async def __call__(self) -> Client: ...


# Correct — sync dependency
def get_client(ctx: DeviceContext) -> Client:
    return ctx.adapter(ClientPort)

Adapter __aenter__ Not Callable¶

Ensure the adapter is a proper async context manager with a callable __aenter__ method.

Negative or Zero Intervals¶

Pass a positive value:

# Wrong
app = App(heartbeat_interval=-5)  # ValueError
app = App(heartbeat_interval=0)  # ValueError

# Correct
app = App(heartbeat_interval=30)

Duplicate Registration¶

Use distinct names for each device. Register only one adapter per port type:

@app.device(name="temperature")
async def temp_device(ctx: DeviceContext) -> dict[str, object]: ...


@app.device(name="humidity")  # Different name
async def humidity_device(ctx: DeviceContext) -> dict[str, object]: ...

Invalid Adapter Tuple¶

Pass either a single adapter instance or a (impl, dry_run) 2-tuple:

# Single adapter
app = App(adapters={MqttPort: my_mqtt_client})

# Adapter + dry-run pair
app = App(adapters={MqttPort: (my_mqtt_client, null_mqtt_client)})

Empty Group Name¶

Pass a non-empty string for the group parameter on @app.device or @app.command.

Persist Without Store¶

Either remove store=None (to use the auto-resolved default store), or pass a store backend explicitly:

from cosalette import App, MemoryStore, SaveOnPublish

# Option A: use the auto-resolved default store (omit store=)
app = App(name="myapp", version="1.0.0")

# Option B: pass an explicit store
app = App(name="myapp", version="1.0.0", store=MemoryStore())


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

Every() — Mutual Exclusion¶

Specify exactly one of seconds or n:

# Wrong
Every(seconds=5, n=10)  # ValueError — both specified
Every()  # ValueError — neither specified

# Correct
Every(seconds=5)
Every(n=10)

Composite Policy and Strategy Children¶

AnySavePolicy, AllSavePolicy, AnyStrategy, and AllStrategy all require at least one child argument. Pass at least one child policy or strategy.

Import Path Format¶

Use the module.path:attr_name colon-separated format: "mypackage.module:MyClass".

Unresolved Provider (Runtime)¶

Register an implementation for the port before the app starts:

app.adapter(SensorPort, Bme280Adapter)

Awaitable Dependency Result (Runtime)¶

Return a plain value from a Depends() callable and await inside the handler instead.

Circular Dependency (Runtime)¶

Break the cycle — extract the shared value into a third dependency that neither side depends on.

Topic Marker and Message Type (Runtime)¶

Topic() and Message require an active MQTT request context:

• Use Topic() only in @app.command handlers or triggered telemetry that is certain to have an incoming message.
• Use Message only in @app.command handlers where an inbound MQTT message is guaranteed.

For scheduled (non-triggered) telemetry, guard with a conditional or restructure to pass the topic via a different mechanism.

Settings Unavailable (Runtime)¶

Set the required environment variables before running the app, or use app.cli() with --env-file to load them from a file.

aiomqtt Not Installed (Runtime)¶

Install the MQTT extra:

pip install cosalette[mqtt]
# or
uv add cosalette[mqtt]

Adapter Not Found (Runtime)¶

Register the required adapter when constructing the app:

app = App(adapters={MqttPort: my_mqtt_client})

Customizing Error Classification¶

The rest of this guide covers how to use build_error_payload() to create domain-specific error types for your own exception handling code. The framework's built-in error isolation publishes your exceptions with error_type="error" — custom classification lets you distinguish error categories in downstream subscribers. Errors the framework raises itself already carry a label (see Framework Error Types), for example invalid_payload for rejected typed-command input and typed telemetry trigger input.

Migrating to 0.12.0 disclosure¶

Before 0.12.0, mapping a type in error_type_map also published its message, and so did the framework's own errors. Both now publish the class name unless you list the type. List everything whose message you still want published: start from the framework's own sanitised errors, then add the app types you mapped before:

# Before 0.12.0: mapping implied disclosure
app = cosalette.App(
    name="wallpanel",
    error_type_map={SshCommandError: "ssh_command"},
)

# 0.12.0: labels and disclosure are separate decisions
app = cosalette.App(
    name="wallpanel",
    error_type_map={SshCommandError: "ssh_command"},
    disclose_messages_for=cosalette.FRAMEWORK_DISCLOSED_ERRORS | {SshCommandError},
)

FRAMEWORK_DISCLOSED_ERRORS leaves out the builtin TimeoutError, so watchdog timeouts publish "TimeoutError" as the message unless you add it yourself.

MQTT__ERROR_PUBLISH_VERBOSE=true alone no longer publishes every message: the app must also pass App(allow_error_publish_verbose=True). Otherwise the setting is ignored with one startup warning.

How Error Publication Works¶

When a device function (telemetry or command) raises an exception:

1. The framework catches it (except CancelledError).
2. Builds a structured ErrorPayload.
3. Publishes to two MQTT topics:
   • {prefix}/error — global error topic (all errors from all devices)
   • {prefix}/{device}/error — per-device error topic
4. Continues running — the error is fire-and-forget. Publication failures are logged but never propagated.

Exception raised in "counter" device
    ↓
Framework catches it, builds ErrorPayload (error_type="error")
    ↓
Publish to gas2mqtt/error         (not retained, QoS 1)
Publish to gas2mqtt/counter/error (not retained, QoS 1)
    ↓
Device loop continues

The ErrorPayload Structure¶

Every error is published as a JSON object with this schema:

{
    "error_type": "sensor_timeout",
    "message": "SensorTimeoutError",
    "device": "counter",
    "timestamp": "2026-02-18T10:30:00+00:00",
    "details": {}
}

Step 1: Define Domain Exceptions¶

Create exception classes for your domain errors:

"""Domain exceptions for gas2mqtt."""


class SensorTimeoutError(Exception):
    """Raised when the gas meter sensor doesn't respond in time."""


class InvalidReadingError(Exception):
    """Raised when a sensor reading is outside valid bounds."""


class CalibrationError(Exception):
    """Raised when the sensor reports calibration failure."""

Step 2: Build the Error Type Map¶

The error type map is a dict mapping exception classes to machine-readable strings:

error_type_map: dict[type[Exception], str] = {
    SensorTimeoutError: "sensor_timeout",
    InvalidReadingError: "invalid_reading",
    CalibrationError: "calibration_error",
}

Step 3: Use build_error_payload()¶

The build_error_payload() function converts an exception into an ErrorPayload:

from cosalette import build_error_payload

error = SensorTimeoutError("Serial read timed out after 5s")

payload = build_error_payload(
    error,
    error_type_map=error_type_map,
    device="counter",
)

print(payload.error_type)  # "sensor_timeout"
print(payload.message)  # "Serial read timed out after 5s"
print(payload.device)  # "counter"
print(payload.to_json())  # Full JSON string

Function Signature¶

def build_error_payload(
    error: Exception,
    *,
    error_type_map: dict[type[Exception], str] | None = None,
    device: str | None = None,
    details: dict[str, object] | None = None,
    clock: Callable[[], datetime] | None = None,
) -> ErrorPayload: ...

Default Fallback¶

Unmapped exceptions get the generic "error" type:

payload = build_error_payload(
    RuntimeError("something broke"),
    error_type_map=error_type_map,
    device="counter",
)

print(payload.error_type)  # "error" — not in the map, so falls back

This ensures every exception produces a valid payload — no exception is ever silently dropped.

Dual Publication¶

The framework publishes each error to two topics:

Errors are published with retain=False and qos=1:

• Not retained because errors are events, not last-known state. You don't want a stale error payload lingering as the retained message.
• QoS 1 (at-least-once) for reliability — error reports should reach subscribers.

Fire-and-Forget Semantics¶

Error publication never crashes the daemon. If publishing itself fails (e.g. MQTT broker is down), the failure is logged but not propagated:

# If the MQTT publish fails:
ERROR    Failed to publish error to gas2mqtt/error
# But the device loop continues — the daemon stays up

This is a deliberate design choice (ADR-011): error reporting must never be the cause of a daemon crash. The framework wraps the entire build → serialise → publish pipeline in a try/except.

Practical Example: Gas Meter Error Types¶

A complete example with domain exceptions for a gas meter bridge:

"""Domain exceptions and error type map for gas2mqtt."""


class SensorTimeoutError(Exception):
    """Gas meter sensor didn't respond within the timeout period."""


class InvalidReadingError(Exception):
    """Sensor returned a reading outside valid physical bounds."""


class ConnectionLostError(Exception):
    """Serial connection to the gas meter was lost."""


# Machine-readable error classification
error_type_map: dict[type[Exception], str] = {
    SensorTimeoutError: "sensor_timeout",
    InvalidReadingError: "invalid_reading",
    ConnectionLostError: "connection_lost",
}

"""gas2mqtt — telemetry device with custom error types."""

import cosalette
from gas2mqtt.errors import InvalidReadingError, SensorTimeoutError
from gas2mqtt.ports import GasMeterPort

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


@app.telemetry("counter", interval=60)
async def counter(ctx: cosalette.DeviceContext) -> dict[str, object]:
    meter = ctx.adapter(GasMeterPort)

    try:
        impulses = meter.read_impulses()
    except TimeoutError as exc:
        raise SensorTimeoutError(f"Read timed out: {exc}") from exc  # (1)!

    if impulses < 0:
        raise InvalidReadingError(  # (2)!
            f"Negative impulse count: {impulses}"
        )

    return {"impulses": impulses}


app.run()

1. Wrap low-level exceptions in domain exceptions. When the framework catches these, it publishes with the generic "error" type and the domain exception class name in the message field; the full text is logged locally.
2. Validate readings and raise domain exceptions for invalid data. The framework catches these, publishes the error, and continues the polling loop.

The framework's automatic error publication produces payloads like:

{
    "error_type": "error",
    "message": "SensorTimeoutError",
    "device": "counter",
    "timestamp": "2026-02-18T10:30:00+00:00",
    "details": {}
}

To get domain-specific error_type values, use build_error_payload() with your error type map in a manual try/except — see Step 3 above for the API.

{
    "error_type": "sensor_timeout",
    "message": "SensorTimeoutError",
    "device": "counter",
    "timestamp": "2026-02-18T10:30:00+00:00",
    "details": {}
}

{
    "error_type": "error",
    "message": "SensorTimeoutError",
    "device": "counter",
    "timestamp": "2026-02-18T10:30:00+00:00",
    "details": {}
}

{
    "error_type": "error",
    "message": "RuntimeError",
    "device": "counter",
    "timestamp": "2026-02-18T10:30:10+00:00",
    "details": {}
}

Testing Error Payloads¶

Test your error type map with plain unit tests:

"""Unit tests for gas2mqtt error types.

Test Techniques Used:
- Decision Table: Exception class → error_type string mapping.
- Specification-based: Verify ErrorPayload structure.
"""

from cosalette import build_error_payload
from gas2mqtt.errors import (
    InvalidReadingError,
    SensorTimeoutError,
    error_type_map,
)


def test_sensor_timeout_maps_correctly():
    """SensorTimeoutError maps to 'sensor_timeout'."""
    payload = build_error_payload(
        SensorTimeoutError("timed out"),
        error_type_map=error_type_map,
        device="counter",
    )

    assert payload.error_type == "sensor_timeout"
    assert payload.device == "counter"
    assert "timed out" in payload.message


def test_unmapped_exception_falls_back():
    """Unmapped exceptions get the default 'error' type."""
    payload = build_error_payload(
        RuntimeError("unexpected"),
        error_type_map=error_type_map,
    )

    assert payload.error_type == "error"


def test_error_payload_serialises_to_json():
    """ErrorPayload.to_json() produces valid JSON."""
    import json

    payload = build_error_payload(
        InvalidReadingError("bad value"),
        error_type_map=error_type_map,
        device="counter",
    )

    data = json.loads(payload.to_json())
    assert data["error_type"] == "invalid_reading"
    assert data["device"] == "counter"

See Also¶

• Error Taxonomy — complete catalog of all framework errors (message text, location, cause)
• Error Handling — conceptual overview of the error publication system
• MQTT Topics — topic layout for error channels
• ADR-011 — error handling and publishing decisions