Error Taxonomy¶

Complete catalog of all exceptions raised by the cosalette framework. Errors are grouped by when they occur: during registration (app setup), at runtime (after bootstrap), or through the error publishing pipeline.

Registration-Time Errors¶

These exceptions are raised before the app starts running — during device registration, decorator application, or App() construction. They indicate a programming mistake in your setup code.

TypeError¶

Registration TypeError exceptions mean the framework received a value of the wrong type, or a callback/handler violates a structural requirement.

Decorator Parentheses¶

Raised when @app.device or @app.command is used without parentheses.

Cause: Python calls the decorator with the function as the first argument when parentheses are missing, which is never the intended use.

Fix: Always use parentheses, 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¶

Raised when the init= parameter of @app.device() receives an async function or a callable with an async __call__.

Cause: The init= callback runs during synchronous bootstrap. Async functions cannot be awaited in that phase.

Fix: Use a regular synchronous function:

# 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¶

Raised when the init= callback returns a type that the framework already provides via dependency injection (e.g. AppContext, MqttPort).

Fix: Wrap the value in a domain-specific type instead of returning a framework type directly.

Bool Parameters (Type Guard)¶

Because bool is a subclass of int (and float), numeric constructor parameters explicitly reject booleans to catch accidental True/False arguments.

Cause: isinstance(True, int) is True in Python, so without an explicit guard, Pt1Filter(tau=True) would silently pass as tau=1.

Fix: Pass a numeric literal:

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

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

Non-Int window¶

Raised when MedianFilter(window=...) receives a value that is not an int (and not a bool).

Fix: Pass an integer: MedianFilter(window=5).

Handler Annotation Errors¶

Raised when a device or command handler has parameters that the injection system cannot resolve.

Cause: The injection system resolves handler parameters by their type annotations. Every parameter must have a concrete type annotation — no *args, **kwargs, positional-only, or non-type annotations.

Fix: Annotate every parameter with a concrete type:

# 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]:
    ...

Adapter __aenter__ Not Callable¶

Raised when an adapter has an __aenter__ attribute that is not callable.

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

Unresolved Interval¶

Raised when a telemetry device's interval is still a callable at runtime, meaning resolve_intervals() was never called during bootstrap.

Cause: This is an internal consistency error — the framework should resolve deferred intervals during bootstrap. If you see this, it may indicate a framework bug or manual misuse of internal APIs.

ValueError¶

Registration ValueError exceptions indicate a value that is the right type but violates a constraint (negative interval, duplicate name, etc.).

Negative or Zero Intervals¶

Raised when a time interval is not positive.

Fix: Pass a positive numeric value:

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

# Correct
app = App(heartbeat_interval=30)

Duplicate Registration¶

Raised when registering a device name or handler that already exists.

Cause: Each device name must be unique within an app (see ADR-019 for scoped name uniqueness). The root device (unnamed) is limited to one.

Fix: Use distinct names for each device:

@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]:
    ...

Duplicate Adapter¶

Raised when registering a second adapter for the same port type.

Fix: Register only one adapter per port type.

Invalid Adapter Tuple¶

Raised when the adapters= dict value is a tuple that is not a 2-tuple.

Fix: Pass either a single adapter instance or a (impl, dry_run) pair:

# 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¶

Raised when a coalescing group name is an empty string.

Fix: Pass a non-empty string for the group parameter.

Persist Without Store¶

Raised when persist=True is set but no store= backend was provided to App().

Fix: Pass a store backend when constructing the app:

from cosalette import App, MemoryStore

app = App(store=MemoryStore())

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

Filter and Strategy Parameters¶

Numeric parameters on filters and strategies must be within valid ranges.

Strategy Mutual Exclusion¶

Every() requires exactly one of seconds or n, not both and not neither.

Fix:

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

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

Composite Policy Children¶

AnySavePolicy and AllSavePolicy require at least one child policy.

Fix: Pass at least one child policy to the composite.

Composite Strategy Children¶

AnyStrategy and AllStrategy require at least one child strategy.

Fix: Pass at least one child strategy to the composite.

Import Path Format¶

Raised when an import path string does not follow the module.path:attr_name convention.

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

Runtime Errors¶

These exceptions are raised after the app has started — during bootstrap completion, MQTT operations, or store access.

RuntimeError¶

Settings Unavailable¶

Raised when the settings model cannot be instantiated, typically because required environment variables are missing.

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

MQTT Not Connected¶

Raised when attempting to publish or subscribe but the MQTT client is not connected.

Cause: Publishing was attempted before the MQTT client connected, or after it disconnected. The framework manages connection lifecycle automatically — this typically indicates use of the MqttClient outside the normal lifecycle.

aiomqtt Not Installed¶

Raised when MqttClient is instantiated but the aiomqtt package is not available.

Fix: Install the MQTT extra: pip install cosalette[mqtt] or uv add cosalette[mqtt].

Store Not Loaded¶

Raised when accessing DeviceStore data before load() has been called.

Cause: The framework calls load() during bootstrap. This error indicates manual use of DeviceStore outside the normal lifecycle, or a framework bug.

Store Not Set¶

Raised internally when create_device_store() is called but no store backend was configured.

LookupError¶

Adapter Not Found¶

Raised when requesting an adapter for a port type that was never registered.

Fix: Register the required adapter when constructing the app:

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

CLI Errors¶

These exceptions are raised by the CLI layer (Typer) when the user provides invalid command-line arguments.

typer.BadParameter¶

SystemExit¶

The CLI exits with code 1 (EXIT_CONFIG_ERROR) when the configuration model raises a validation error (e.g. from pydantic).

Error Publishing Pipeline¶

The framework includes a built-in error publishing system for reporting runtime errors via MQTT. This section summarizes the pipeline — for full details, see Error Handling (concept) and Payload Schemas.

Pipeline Flow¶

Exception raised in device function
    ↓
build_error_payload(error, error_type_map=..., device=...)
    ↓
ErrorPayload(error_type, message, device, timestamp, details)
    ↓
ErrorPublisher.publish()
    ↓
MQTT: {prefix}/error              (global, always)
MQTT: {prefix}/{device}/error     (per-device, when device known)

error_type_map Pattern¶

The error_type_map is a dict[type[Exception], str] that maps exact exception classes (no subclass matching) to machine-readable error_type strings. Unmapped exceptions produce "error" as the type.

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

Publication Behaviour¶

Topic Layout¶

Root devices (unnamed) only publish to the global topic to avoid duplicating the same error on both topics.

See Also¶

• Error Handling (concept) — design principles and rationale
• Payload Schemas — JSON schema for ErrorPayload
• Map Custom Error Types (guide) — how to use build_error_payload() with custom domain exceptions
• ADR-011 — architecture decision record