API Reference¶

def __init__(
    self,
    name: str,
    version: str = "0.0.0",
    *,
    description: str = "IoT-to-MQTT bridge",
    settings_class: type[Settings] = Settings,
    dry_run: bool = False,
    heartbeat_interval: float | None = 60.0,
    heartbeat_include_version: bool = True,
    health_check_interval: float | None = 30.0,
    lifespan: LifespanFunc | None = None,
    store: Store | Callable[..., Store] | None | _Unset = _UNSET,
    retained_cleanup: bool | None = None,
    retained_cleanup_snapshot_key: SecretStr | None = None,
    adapters: dict[
        type,
        type
        | str
        | Callable[..., object]
        | tuple[
            type | str | Callable[..., object],
            type | str | Callable[..., object],
        ],
    ]
    | None = None,
    restart_after_failures: int = 5,
    max_restarts: int = 3,
    restart_cooldown: float = 5.0,
    sustained_health_reset: float = 300.0,
    error_type_map: dict[type[Exception], str] | None = None,
    disclose_messages_for: frozenset[type[Exception]] | None = None,
) -> None:
    """Initialise the application orchestrator.

    Args:
        name: Application name (used as MQTT topic prefix and client ID).
        version: Application version string.
        description: Short description for CLI help text.
        settings_class: Settings subclass to instantiate at startup.
        dry_run: When True, resolve dry-run adapter variants.
        heartbeat_interval: Seconds between periodic heartbeats
            published to ``{prefix}/status``.  Set to ``None`` to
            disable periodic heartbeats entirely.  Defaults to 60.
        heartbeat_include_version: When ``False``, omit the
            ``version`` field from the heartbeat JSON (F-DP6).
            Broker observers could otherwise match the disclosed
            version against published CVEs.  The version remains in
            local logs and the package metadata.  Defaults to True.
        health_check_interval: Seconds between periodic health
            checks for adapters implementing
            :class:`~cosalette.HealthCheckable`.  Set to ``None`` to
            disable health checks entirely.  Defaults to 30.
        lifespan: Async context manager for application startup
            and shutdown.  Code before ``yield`` runs before devices
            start; code after ``yield`` runs after devices stop.
            Receives an :class:`AppContext`.  When ``None``, a no-op
            default is used.
        store: Persistence backend for device state.  When omitted,
            the framework auto-creates a :class:`JsonFileStore` at
            a default path derived from *name* — see ADR-049 for
            the full resolution order (``<NAME>_STORE_PATH`` env,
            then ``$XDG_STATE_HOME``, then ``~/.local/state``).
            Pass ``store=None`` to opt out of all persistence.
            Pass an explicit :class:`Store` instance or a
            ``Callable[..., Store]`` factory to override.
        retained_cleanup: Tri-state override for ADR-048 retained-topic
            cleanup.  ``None`` (default) leaves the auto-heuristic in
            charge: cleanup runs when the entity set may vary by config
            or an explicit ``store=`` was passed (see
            :attr:`has_dynamic_entities`).  ``False`` disables cleanup
            entirely — the cleanup store resolves to ``None`` so no
            snapshot I/O runs — and suppresses the ephemeral-store
            warning for an auto-default store on an ephemeral filesystem.
            ``True`` forces cleanup on and the ephemeral-store warning
            fires for an auto-default store on an ephemeral filesystem,
            even for provably-static apps.  When combined with
            ``store=None``, ``True`` is a graceful no-op (there is no
            store to hold the ADR-048 snapshot).
            See ADR-048 (orphaned retained-topic cleanup) and ADR-049
            (default store path resolution and the opt-out).
        retained_cleanup_snapshot_key: Opt-in HMAC-SHA256 signing key for
            the ADR-048 retained-cleanup entity snapshot (F-DP3).
            ``None`` (default) preserves today's unsigned behavior
            exactly — no behavior change. When supplied, the persisted
            snapshot is signed on save and verified (timing-safe) on
            load; a missing, unrecognized-algorithm, or mismatched
            signature — including a pre-existing *unsigned* snapshot
            from before a key was configured — is treated as no
            previous snapshot (fail-closed): this run's cleanup is
            skipped and a freshly signed snapshot is written. Signing
            detects tampering by a writer with access to the `Store`
            backend; it does **not** address the module's documented
            single-writer concurrency race. Source the key's value the
            same way as ``MqttSettings.password`` (env var, secrets
            manager, etc.) — never via the same `Store` this feature
            protects. See ADR-063.
        adapters: Optional mapping of port types to adapter
            implementations.  Each key is a Protocol type; each
            value is either a single implementation (class,
            lazy-import string, or factory callable) or a
            ``(impl, dry_run)`` tuple.  Entries are registered via
            :meth:`adapter` and coexist with later imperative calls.
        error_type_map: Optional mapping from app-owned exception types to
            machine-readable ``error_type`` strings.  Registering a type
            opts it back into full-message error publishing under the
            LEAK-01 default-deny: an exception whose type is in the map has
            its ``str(error)`` published on the broker-visible error topic,
            while unlisted (downstream/unexpected) exceptions keep having
            their message redacted to the class name.  Framework command
            exceptions remain authoritative and cannot be overridden.  See
            ADR-011.

            Security note: mapping a type is a **message-disclosure
            decision**, not just labeling.  Only register exceptions whose
            messages are guaranteed free of secrets, filesystem paths,
            hostnames, or credentials — exception text frequently embeds
            URLs with userinfo or absolute paths.
        disclose_messages_for: Explicit set of exception types whose
            ``str(error)`` may be published, decoupled from
            ``error_type_map`` labeling (F-DP1).  When given, it **fully
            defines** the disclosure policy — a type must be listed here
            to have its message published, regardless of whether it is
            also in ``error_type_map`` (framework-mapped types included).
            ``None`` (default) preserves the legacy conflated behaviour:
            registering a type in ``error_type_map`` alone discloses its
            message.  See ADR-061.
    """
    validate_mqtt_name(name)
    if not name.strip():
        # The app name is the MQTT topic root prefix and is emitted as the
        # channel-level ``x-cosalette-app`` tag (ADR-033).  An empty/blank
        # name would yield ``/device/state`` topics and an ``x-cosalette-app:
        # ''`` that the schema loader rejects on read-back — breaking schema
        # enforcement closed.  Reject it at the source.
        msg = f"App name must be a non-empty, non-blank string, got {name!r}"
        raise ValueError(msg)
    self._name = name
    self._version = version
    self._description = description
    self._settings_class = settings_class
    try:
        self._settings: Settings | None = settings_class()
    except ValidationError, SettingsLoadError:
        self._settings = None
    self._dry_run = dry_run
    _validate_positive_interval("heartbeat_interval", heartbeat_interval)
    self._heartbeat_interval = heartbeat_interval
    self._heartbeat_include_version = heartbeat_include_version
    _validate_positive_interval("health_check_interval", health_check_interval)
    self._health_check_interval = health_check_interval
    if restart_after_failures < 0:
        msg = f"restart_after_failures must be >= 0, got {restart_after_failures}"
        raise ValueError(msg)
    self._restart_after_failures = restart_after_failures
    if max_restarts < 0:
        msg = f"max_restarts must be >= 0, got {max_restarts}"
        raise ValueError(msg)
    self._max_restarts = max_restarts
    _validate_positive_interval("restart_cooldown", restart_cooldown)
    self._restart_cooldown = restart_cooldown
    _validate_positive_interval("sustained_health_reset", sustained_health_reset)
    self._sustained_health_reset = sustained_health_reset
    self._lifespan: LifespanFunc = (
        lifespan if lifespan is not None else _noop_lifespan
    )
    self._devices: list[_DeviceRegistration] = []
    self._telemetry: list[_TelemetryRegistration] = []
    self._commands: list[_CommandRegistration] = []
    self._streams: list[_StreamRegistration] = []
    self._periodic: list[_PeriodicRegistration] = []
    self._reactors: list[_ReactorRegistration] = []
    self._state_factories: list[StateRegistration] = []
    self._state_overrides: dict[type, Any] = {}  # for tests
    self._adapters: dict[type, _AdapterEntry] = {}
    self._store_factory: Callable[..., Store] | None = None
    self._store: Store | None = None
    self._store_is_default = False
    self._entity_set_is_dynamic: bool | None = None
    self._retained_cleanup = retained_cleanup
    self._retained_cleanup_snapshot_key = retained_cleanup_snapshot_key
    self._discovery: DiscoveryConfig | None = None
    self._error_type_map = _validate_error_type_map(error_type_map)
    self._disclose_messages_for = _validate_disclose_messages_for(
        disclose_messages_for
    )
    self._apply_store_arg(store)
    self._configure_hooks: list[Callable[..., Any]] = []

    def _register(
        pt: type,
        impl: type | str | Callable[..., object],
        dry_run: type | str | Callable[..., object] | None,
    ) -> None:
        self.adapter(pt, impl, dry_run=dry_run)

    process_adapters_dict(adapters, _register)

def __init__(
    self,
    *,
    settings: Settings,
    adapters: dict[type, object],
) -> None:
    """Initialise lifecycle-hook context.

    Args:
        settings: Application settings instance.
        adapters: Resolved adapter registry mapping port types to instances.
    """
    self._settings = settings
    self._adapters = adapters

def __init__(
    self,
    *,
    name: str,
    settings: Settings,
    mqtt: MqttPort,
    topic_prefix: str,
    shutdown_event: asyncio.Event,
    adapters: dict[type, object],
    clock: ClockPort,
    is_root: bool = False,
    health_reporter: HealthReporter | None = None,
    state_model: type | None = None,
    handler_name: str | None = None,
    command_maxsize: int = 0,
    command_backpressure: BackpressurePolicy = "drop_newest",
) -> None:
    """Initialise per-device context.

    Args:
        name: Device name as registered (e.g. "blind").
        settings: Application settings instance.
        mqtt: MQTT port for publishing.
        topic_prefix: Root prefix for MQTT topics (e.g. "velux2mqtt").
        shutdown_event: Shared event that signals graceful shutdown.
        adapters: Resolved adapter registry mapping port types to instances.
        clock: Monotonic clock for timing.
        is_root: When True, topics omit the device name segment
            (root-level device).
        health_reporter: Optional health reporter for availability
            signalling.
        state_model: Declared state contract for this handler's static
            ``state`` topic.  When set, every :meth:`publish_state`
            payload is validated and normalised against it.  ``None``
            (default) skips validation entirely — no ``TypeAdapter`` is
            built and no per-publish cost is added.
        handler_name: Qualified handler name, used only to make
            validation errors actionable.
        command_maxsize: Maximum size for the device command queue.
            ``0`` (default) means unbounded.
        command_backpressure: Policy applied when ``command_maxsize > 0``
            and the command queue is full.

    See Also:
        ADR-045 (amended 2026-08-07) — ``state_model`` is runtime
        load-bearing for ``@app.stream`` and ``@app.device``.
    """
    self._name = name
    self._settings = settings
    self._mqtt = mqtt
    self._topic_prefix = topic_prefix
    self._shutdown_event = shutdown_event
    self._adapters = adapters
    self._clock = clock
    self._command_handlers: dict[str | None, CommandHandler] = {}
    self._command_timeouts: dict[str | None, float | None] = {}
    self._is_root = is_root
    self._command_backpressure = command_backpressure
    self._log_label = f"command for {name!r}"
    self._command_queue: asyncio.Queue[Command] = asyncio.Queue(
        maxsize=command_maxsize
    )
    self._commands_consumed: bool = False
    self._topic_base = topic_prefix if is_root else f"{topic_prefix}/{name}"
    self._active_sub_entities: set[str] = set()
    self._health_reporter = health_reporter
    self._is_unavailable: bool = False
    self._state_model = state_model
    self._handler_name = handler_name