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,
    startup_connect_timeout: float | None = 10.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,
    allow_error_publish_verbose: bool = False,
    on_task_failure: TaskFailurePolicy = "restart",
    task_max_restarts: int = DEFAULT_TASK_MAX_RESTARTS,
    task_restart_window: float = DEFAULT_TASK_RESTART_WINDOW,
    error_reminder_interval: float | None = DEFAULT_ERROR_REMINDER_INTERVAL,
    exit_after_stale: float | None = None,
    restart_on_stale: bool = False,
    redact: RedactSpec = 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.
        startup_connect_timeout: Upper bound, in seconds, on how long
            device, telemetry, periodic and stream tasks wait at startup
            for the first MQTT connection (and its availability
            announce) before they run.  A broker that is still
            unreachable afterwards logs one warning; handlers then run.
            Telemetry and command context publishes raise
            :class:`MqttNotConnectedError` until it connects (telemetry
            skips them without an error report), while ``@app.device``
            and ``@app.stream`` handler publishes are dropped at
            ``DEBUG`` so their generators keep running. Set to ``None``
            to start handlers immediately. Only
            applies to connect-aware MQTT adapters such as
            :class:`MqttClient`.  Defaults to 10.
        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.  It only assigns
            labels; it never publishes a message (see
            *disclose_messages_for*).  Framework labels
            (``invalid_json``, ``missing_sub_key``,
            ``unknown_sub_command``, ``invalid_payload`` for
            :class:`PayloadValidationError`, ``timeout``) remain
            authoritative; an entry overriding one with a different label
            logs a startup warning.  See ADR-011 and ADR-091.
        disclose_messages_for: The exception types whose ``str(error)``
            is published on the broker-visible error topics (F-DP1),
            exact classes only.  ``None`` (default) discloses nothing:
            every error, the framework's own included, publishes its
            ``error_type`` label and the exception class name as
            ``message``.  Pass
            ``cosalette.FRAMEWORK_DISCLOSED_ERRORS | {MyError}`` to
            publish the framework's sanitised command error messages
            plus your own (the builtin ``TimeoutError`` is disclosed only
            when listed).  Only list exceptions whose messages are free
            of secrets, filesystem paths, hostnames or credentials.  See
            ADR-091.
        allow_error_publish_verbose: Allow verbose error publishing,
            which discloses the message of **every** error.  It takes
            effect only when the operator also sets
            ``MqttSettings.error_publish_verbose``
            (``MQTT__ERROR_PUBLISH_VERBOSE=true``); if the setting is on
            and this is ``False``, the setting is ignored and one WARNING
            is logged at startup.  *redact* still applies to every
            disclosed message.  Defaults to False.  See ADR-091.
        on_task_failure: What happens when a framework-started task
            (device, telemetry entity, coalescing group, periodic or
            stream handler) dies.  ``"restart"`` (default) re-creates it
            with exponential backoff (1 s doubling to 60 s) within the
            restart budget and exits with code 4 once the budget is
            exhausted; ``"exit"`` exits with code 4 on the first
            failure; ``"ignore"`` leaves the entity offline with status
            ``error``.  Every policy logs the failure, publishes one
            error payload and marks the entities offline.  See ADR-081.
        task_max_restarts: Restarts allowed per registration (per group
            for a coalescing group) before the app exits.  ``0`` makes
            ``"restart"`` exit on the first failure.
        task_restart_window: Seconds without a failure after which a
            registration's restart count resets to 0.
        error_reminder_interval: Seconds between reminders for a
            telemetry failure that persists.  The first error of each
            type is logged and published; while the same error repeats,
            the 2nd, 4th, 8th, ... failure within the first interval and
            then one failure per interval are logged at WARNING and
            republished with ``details.count`` and
            ``details.first_seen``.  ``None`` disables reminders, leaving
            only the onset and the recovery line.  Defaults to 3600.
            See ADR-082.
        exit_after_stale: Seconds a telemetry entity or stream may stay ``stale``
            (ADR-080) before the app logs CRITICAL and shuts down with
            exit code 5, so a container restart policy can recover it.
            Counted from the moment the entity went stale.  ``None``
            (default) never exits.  With *restart_on_stale*, keep it
            above the in-place recovery time, or the app exits first:
            ``exit_after_stale > check_interval + restart_time +
            first_cycle_time``, where ``check_interval`` is
            ``min(heartbeat_interval, 60 s, smallest stale_after)``,
            ``restart_time`` is *restart_cooldown* plus ``reset()`` or
            re-entry plus the health check after it, and
            ``first_cycle_time`` is the first successful cycle of the
            recreated telemetry.  Rule of thumb: at least
            ``2 * (60 + restart_cooldown + longest telemetry interval)``.
            See ADR-083.
        restart_on_stale: When ``True``, a telemetry entity or a
            stream with ``stale_after=`` that goes ``stale`` requests a
            restart of every restartable adapter it depends on, without
            waiting for failed health checks.
            Each request counts against *max_restarts* and fires once
            per stale episode.  Needs *health_check_interval*.
            Defaults to False.  See ADR-084.
        redact: Scrubs secrets from text that leaves the process: the
            message of a disclosed error payload (``disclose_messages_for``
            or verbose publishing)
            and every record on the log handlers cosalette installs.
            ``None`` (default) is off; a callable ``str -> str`` is used
            as is; an iterable of regular expressions (``str`` or
            compiled) replaces each match with ``[REDACTED]``.  A
            redactor that raises lets the text through unchanged and logs
            one WARNING.  Undisclosed errors still publish only the class
            name.  See ADR-085.

    Raises:
        TypeError: If *redact* is not ``None``, a callable or an
            iterable of ``str`` / ``re.Pattern``.
        ValueError: If a *redact* pattern is not a valid regular
            expression.
    """
    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
    _validate_positive_interval("startup_connect_timeout", startup_connect_timeout)
    self._startup_connect_timeout = startup_connect_timeout
    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._on_task_failure = validate_task_failure_policy(on_task_failure)
    if (
        isinstance(task_max_restarts, bool)
        or not isinstance(task_max_restarts, int)
        or task_max_restarts < 0
    ):
        msg = f"task_max_restarts must be an int >= 0, got {task_max_restarts!r}"
        raise ValueError(msg)
    self._task_max_restarts = task_max_restarts
    _validate_positive_interval("task_restart_window", task_restart_window)
    self._task_restart_window = task_restart_window
    _validate_positive_interval("error_reminder_interval", error_reminder_interval)
    self._error_reminder_interval = error_reminder_interval
    _validate_positive_interval("exit_after_stale", exit_after_stale)
    self._exit_after_stale = exit_after_stale
    self._restart_on_stale = _validate_bool_option(
        "restart_on_stale", restart_on_stale
    )
    self._allow_error_publish_verbose = _validate_bool_option(
        "allow_error_publish_verbose", allow_error_publish_verbose
    )
    self._redactor = build_redactor(redact)
    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._inbounds: list[_InboundRegistration] = []
    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",
    tolerate_not_connected: bool = False,
) -> 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.
        tolerate_not_connected: When True, :meth:`publish_state`,
            :meth:`publish` and sub-entity publishes log
            :class:`~cosalette.MqttNotConnectedError` at ``DEBUG`` and
            return instead of raising.  The framework
            sets it for ``@app.device`` and ``@app.stream`` contexts so a
            broker outage does not end their generators; telemetry and
            command contexts keep raising (ADR-011 amendment,
            2026-10-01).

    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._availability_source = "manual"
    self._state_model = state_model
    self._handler_name = handler_name
    self._tolerate_not_connected = tolerate_not_connected