From 5ed83ae9e3b94b04971a1057add7546e265f6fb3 Mon Sep 17 00:00:00 2001
From: Goffi <goffi@goffi.org>
Date: Thu, 2 Jul 2026 15:35:23 +0200
Subject: [PATCH] plugin XEP-0085: Modernisation:

- Async coroutines, type hints
- Use client plugin data instead of per-profile dict
- Detect chat state availability with disco queries
- Better spec compliance
---
 libervia/backend/plugins/plugin_xep_0085.py | 529 +++++++++++++-------
 1 file changed, 356 insertions(+), 173 deletions(-)

diff --git a/libervia/backend/plugins/plugin_xep_0085.py b/libervia/backend/plugins/plugin_xep_0085.py
index c575eed3..55a2d380 100644
--- a/libervia/backend/plugins/plugin_xep_0085.py
+++ b/libervia/backend/plugins/plugin_xep_0085.py
@@ -1,7 +1,7 @@
 #!/usr/bin/env python3
 
-
-# SAT plugin for Chat State Notifications Protocol (xep-0085)
+# Libervia plugin for Chat State Notifications Protocol (XEP-0085)
+# Copyright (C) 2009-2026 Jérôme Poisson (goffi@goffi.org)
 # Copyright (C) 2009-2016 Adrien Cossa (souliane@mailoo.org)
 
 # This program is free software: you can redistribute it and/or modify
@@ -17,20 +17,32 @@
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
-from libervia.backend.core.i18n import _
-from libervia.backend.core.constants import Const as C
-from libervia.backend.core import exceptions
-from libervia.backend.core.log import getLogger
+from dataclasses import dataclass, field
+from typing import Any, TYPE_CHECKING, cast
 
-log = getLogger(__name__)
-from tx_xmpp import disco, itx_xmpp
-from zope.interface import implementer
-from twisted.words.protocols.jabber.jid import JID
-
-from twisted.words.protocols.jabber.xmlstream import XMPPHandler
-from twisted.words.xish import domish
+from twisted.internet import defer
 from twisted.internet import reactor
 from twisted.internet import error as internet_error
+from twisted.internet.base import DelayedCall
+from twisted.internet.interfaces import IReactorTime
+from twisted.words.protocols.jabber.jid import JID
+from twisted.words.protocols.jabber.xmlstream import XMPPHandler
+from twisted.words.xish import domish
+from tx_xmpp import disco, itx_xmpp
+from zope.interface import implementer
+
+from libervia.backend.core import exceptions
+from libervia.backend.core.constants import Const as C
+from libervia.backend.core.core_types import SatXMPPEntity
+from libervia.backend.core.i18n import _
+from libervia.backend.core.log import getLogger
+
+if TYPE_CHECKING:
+    from libervia.backend.core.main import LiberviaBackend
+
+
+log = getLogger(__name__)
+
 
 NS_XMPP_CLIENT = "jabber:client"
 NS_CHAT_STATES = "http://jabber.org/protocol/chatstates"
@@ -52,6 +64,18 @@ PLUGIN_INFO = {
     C.PI_DESCRIPTION: _("""Implementation of Chat State Notifications Protocol"""),
 }
 
+IMPORT_NAME = PLUGIN_INFO[C.PI_IMPORT_NAME]
+
+
+@dataclass
+class ChatStatesData:
+    """Per-client data for XEP-0085 chat state notifications."""
+
+    # peer_jid -> Allowed (True)/Denied (False)
+    disco_cache: dict[JID, bool] = field(default_factory=dict)
+    # Chat state machines, keyed by peer_jid
+    machines: dict[JID, "ChatStateMachine"] = field(default_factory=dict)
+
 
 # Describe the internal transitions that are triggered
 # by a timer. Beside that, external transitions can be
@@ -74,7 +98,7 @@ class UnknownChatStateException(Exception):
     pass
 
 
-class XEP_0085(object):
+class XEP_0085:
     """
     Implementation for XEP 0085
     """
@@ -94,13 +118,9 @@ class XEP_0085(object):
         "param_label": _("Enable chat state notifications"),
     }
 
-    def __init__(self, host):
+    def __init__(self, host: "LiberviaBackend") -> None:
         log.info(_("Chat State Notifications plugin initialization"))
         self.host = host
-        self.map = (
-            {}
-        )  # FIXME: would be better to use client instead of mapping profile to data
-
         # parameter value is retrieved before each use
         host.memory.update_params(self.params)
 
@@ -121,26 +141,104 @@ class XEP_0085(object):
         # args: from (jid as string), state in CHAT_STATES, profile
         host.bridge.add_signal("chat_state_received", ".plugin", signature="sss")
 
-    def get_handler(self, client):
+    def get_handler(self, client: SatXMPPEntity) -> "XEP_0085_handler":
         return XEP_0085_handler(self, client.profile)
 
-    def profile_disconnected(self, client):
+    def profile_disconnected(self, client: SatXMPPEntity) -> None:
         """Eventually send a 'gone' state to all one2one contacts."""
-        profile = client.profile
-        if profile not in self.map:
-            return
-        for to_jid in self.map[profile]:
+        chat_states_data = client.get_plugin_data(IMPORT_NAME, ChatStatesData)
+        for machine in list(chat_states_data.machines.values()):
             # FIXME: the "unavailable" presence stanza is received by to_jid
             # before the chat state, so it will be ignored... find a way to
             # actually defer the disconnection
-            self.map[profile][to_jid]._onEvent("gone")
-        del self.map[profile]
+            machine._onEvent("gone")
+        chat_states_data.machines.clear()
 
-    def update_cache(self, entity_jid, value, profile):
+    async def _check_disco_support(
+        self, client: SatXMPPEntity, peer_jid: JID
+    ) -> bool | None:
+        """Check if a contact supports XEP-0085 via service discovery.
+
+        For 1:1 (bare JID): checks all known resources.
+        For MUC DM (full JID): checks the occupant directly.
+
+        @param client: the client instance
+        @param peer_jid: bare JID for 1:1, full JID for MUC DM
+        @return: True if confirmed, False if denied, None if no resources known
+        """
+        data = client.get_plugin_data(IMPORT_NAME, ChatStatesData)
+        if peer_jid in data.disco_cache:
+            return data.disco_cache[peer_jid]
+
+        if not peer_jid.resource:
+            # 1:1 (we check all known resources)
+            try:
+                resources = self.host.memory.get_all_resources(client, peer_jid)
+            except exceptions.UnknownEntityError:
+                # entity not in cache
+                return None
+            if not resources:
+                # no resources known yet
+                return None
+            for resource in resources:
+                full_jid = JID(None, (peer_jid.user, peer_jid.host, resource))
+                try:
+                    has_feature = await self.host.memory.disco.has_feature(
+                        client, NS_CHAT_STATES, full_jid
+                    )
+                    if has_feature:
+                        data.disco_cache[peer_jid] = True
+                        return True
+                except (
+                    exceptions.ExternalRequestError,
+                    exceptions.TimeOutError,
+                ) as e:
+                    log.warning(
+                        f"Can't check chat state availability for {full_jid}: {e}"
+                    )
+                    continue
+            data.disco_cache[peer_jid] = False
+            return False
+
+        # We have a full JID -> This is a MUC DM
+        try:
+            has_feature = await self.host.memory.disco.has_feature(
+                client, NS_CHAT_STATES, peer_jid
+            )
+        except (
+            exceptions.ExternalRequestError,
+            exceptions.TimeOutError,
+        ):
+            log.debug(
+                f"Disco query for chat states support failed for {peer_jid.full()}, "
+                "falling back to implicit discovery"
+            )
+            has_feature = False
+        data.disco_cache[peer_jid] = has_feature
+        return has_feature
+
+    async def _ensure_disco_check(self, client: SatXMPPEntity, to_jid: JID) -> None:
+        """Check disco support for chat states if not already done.
+
+        Skipped for MUC groupchat.
+        For 1:1: passes bare JID, checks all known resources.
+        For MUC DM: passes full JID, checks occupant.
+
+        @param client: the client instance
+        @param to_jid: JID of the contact
+        """
+        if self._is_muc(to_jid, client.profile) and not to_jid.resource:
+            # This is a MUC groupchat, we skip
+            return
+        data = client.get_plugin_data(IMPORT_NAME, ChatStatesData)
+        if to_jid not in data.disco_cache:
+            await self._check_disco_support(client, to_jid)
+
+    def update_cache(self, entity_jid: JID, value: bool | str, profile: str) -> None:
         """Update the entity data of the given profile for one or all contacts.
         Reset the chat state(s) display if the notification has been disabled.
 
-        @param entity_jid: contact's JID, or C.ENTITY_ALL to update all contacts.
+        @param entity_jid: contact's JID
         @param value: True, False or DELETE_VALUE to delete the entity data
         @param profile: current profile
         """
@@ -150,108 +248,164 @@ class XEP_0085(object):
         else:
             self.host.memory.update_entity_data(client, entity_jid, ENTITY_KEY, value)
         if not value or value == DELETE_VALUE:
-            # reinit chat state UI for this or these contact(s)
+            # reinit chat state UI for this contact
             self.host.bridge.chat_state_received(entity_jid.full(), "", profile)
 
-    def param_update_trigger(self, name, value, category, type_, profile):
-        """Reset all the existing chat state entity data associated with this profile after a parameter modification.
+    def param_update_trigger(
+        self, name: str, value: str, category: str, type_: str, profile: str
+    ) -> bool:
+        """Update Chat State data according to parameter.
 
-        @param name: parameter name
-        @param value: "true" to activate the notifications, or any other value to delete it
-        @param category: parameter category
-        @param type_: parameter type
+        @param name: Parameter name.
+        @param value: "true" to activate the notifications, or any other value to delete
+            it.
+        @param category: Parameter category.
+        @param type_: Parameter type.
         """
         if (category, name) == (PARAM_KEY, PARAM_NAME):
-            self.update_cache(
-                C.ENTITY_ALL, True if C.bool(value) else DELETE_VALUE, profile=profile
-            )
+            client = self.host.get_client(profile)
+            if C.bool(value):
+                self.host.memory.update_entity_data(
+                    client, C.ENTITY_ALL, ENTITY_KEY, True
+                )
+            else:
+                self.host.memory.del_entity_datum(client, C.ENTITY_ALL, ENTITY_KEY)
             return False
         return True
 
-    def message_received_trigger(self, client, message, post_treat):
-        """
-        Update the entity cache when we receive a message with body.
-        Check for a chat state in the message and signal frontends.
+    async def message_received_trigger(
+        self, client: SatXMPPEntity, message: domish.Element, post_treat: bool
+    ) -> bool:
+        """Handle an incoming message.
+
+        Update entity cache, init state machine, and signal frontends about
+        received chat states.
         """
         profile = client.profile
         if not self.host.memory.param_get_a(PARAM_NAME, PARAM_KEY, profile_key=profile):
             return True
 
         from_jid = JID(message.getAttribute("from"))
-        if self._is_muc(from_jid, profile):
+        msg_type = message.getAttribute("type")
+        # MUC: for groupchat we remove resource to get room bare JID.
+        #      for DM (type="chat") we keep full JID to track occupant.
+        is_groupchat = self._is_muc(from_jid, profile) and msg_type == "groupchat"
+        if is_groupchat:
             from_jid = from_jid.userhostJID()
-        else:  # update entity data for one2one chat
-            # assert from_jid.resource # FIXME: assert doesn't work on normal message from server (e.g. server announce), because there is no resource
-            try:
-                next(domish.generateElementsNamed(message.elements(), name="body"))
-                try:
-                    next(domish.generateElementsNamed(message.elements(), name="active"))
-                    # contact enabled Chat State Notifications
-                    self.update_cache(from_jid, True, profile=profile)
-                except StopIteration:
-                    if message.getAttribute("type") == "chat":
-                        # contact didn't enable Chat State Notifications
-                        self.update_cache(from_jid, False, profile=profile)
-                        return True
-            except StopIteration:
-                pass
 
-        # send our next "composing" states to any MUC and to the contacts who enabled the feature
-        self._chat_state_init(from_jid, message.getAttribute("type"), profile)
+        # Extract any chat state element from the message (for all message types)
+        chat_state = next(
+            (
+                child.name
+                for child in message.elements()
+                if child.name in CHAT_STATES and child.defaultUri == NS_CHAT_STATES
+            ),
+            None,
+        )
 
-        state_list = [
-            child.name
-            for child in message.elements()
-            if message.getAttribute("type") in MESSAGE_TYPES
-            and child.name in CHAT_STATES
-            and child.defaultUri == NS_CHAT_STATES
-        ]
-        for state in state_list:
-            # there must be only one state according to the XEP
-            if state != "gone" or message.getAttribute("type") != "groupchat":
-                self.host.bridge.chat_state_received(
-                    message.getAttribute("from"), state, profile
+        if not is_groupchat:
+            # one2one or MUC DM: implicit discovery (XEP-0085 §5.1)
+            # check if this is a content message (has <body/>)
+            has_body = (
+                next(
+                    domish.generateElementsNamed(message.elements(), name="body"),
+                    None,
+                )
+                is not None
+            )
+
+            if chat_state is not None:
+                # XEP-0085 §5.1 Rule 3: any chat state proves support
+                self.update_cache(from_jid, True, profile=profile)
+            elif has_body and message.getAttribute("type") == "chat":
+                # XEP-0085 §5.1 Rule 2: content message without chat state
+                # only deactivate if not already activated
+                try:
+                    current = self.host.memory.get_entity_datum(
+                        client, from_jid, ENTITY_KEY
+                    )
+                except (exceptions.UnknownEntityError, KeyError):
+                    current = None
+                if not current:
+                    self.update_cache(from_jid, False, profile=profile)
+                    return True
+
+            # fire explicit disco check for this contact
+            await self._ensure_disco_check(client, from_jid)
+
+        # init state machine so composing states can be sent to this contact
+        # XEP-0085 §5.4: only use with chat/groupchat message types
+        if msg_type in MESSAGE_TYPES:
+            self._chat_state_init(client, from_jid, msg_type)
+
+        # Signal frontends about received chat states
+        if chat_state is not None:
+            # XEP-0085 §5.5: ignore <gone/> in groupchat
+            if chat_state != "gone" or message.getAttribute("type") != "groupchat":
+                self.host.bridge.chat_state_received(
+                    message.getAttribute("from"), chat_state, profile
                 )
-            break
         return True
 
     def send_message_trigger(
-        self, client, mess_data, pre_xml_treatments, post_xml_treatments
-    ):
-        """
-        Eventually add the chat state to the message and initiate
-        the state machine when sending an "active" state.
-        """
+        self,
+        client: SatXMPPEntity,
+        mess_data: dict,
+        pre_xml_treatments: "defer.Deferred",
+        post_xml_treatments: "defer.Deferred",
+    ) -> bool:
+        """Add chat state to outgoing messages and initiate the state machine."""
         profile = client.profile
 
-        def treatment(mess_data):
+        async def treatment(mess_data):
             message = mess_data["xml"]
             to_jid = JID(message.getAttribute("to"))
-            if not self._check_activation(to_jid, forceEntityData=True, profile=profile):
+
+            # XEP-0085 §4: fire disco check for this contact
+            await self._ensure_disco_check(client, to_jid)
+
+            if not await self._check_activation(client, to_jid, forceEntityData=True):
                 return mess_data
-            try:
-                # message with a body always mean active state
-                next(domish.generateElementsNamed(message.elements(), name="body"))
+
+            # XEP-0085 §2: check if this is a content message (has <body/>)
+            has_body = (
+                next(
+                    domish.generateElementsNamed(message.elements(), name="body"),
+                    None,
+                )
+                is not None
+            )
+
+            if has_body and mess_data["type"] in MESSAGE_TYPES:
+                # XEP-0085 §5.4: only use with chat/groupchat message types
+                # XEP-0085 §5.6: content message -> attach <active/>
                 message.addElement((NS_CHAT_STATES, "active"))
-                # launch the chat state machine (init the timer)
-                if self._is_muc(to_jid, profile):
-                    to_jid = to_jid.userhostJID()
-                self._chat_state_active(to_jid, mess_data["type"], profile)
-            except StopIteration:
-                if "chat_state" in mess_data["extra"]:
-                    state = mess_data["extra"].pop("chat_state")
-                    assert state in CHAT_STATES
-                    message.addElement((NS_CHAT_STATES, state))
+                # Launch the chat state machine (init the timer)
+                # MUC groupchat: strip resource to get room bare JID.
+                # MUC DM (type="chat"): keep full JID to track occupant.
+                target_jid = (
+                    to_jid.userhostJID()
+                    if self._is_muc(to_jid, profile) and mess_data["type"] == "groupchat"
+                    else to_jid
+                )
+                self._chat_state_active(client, target_jid, mess_data["type"])
+            elif "chat_state" in mess_data["extra"]:
+                # Standalone notification
+                state = mess_data["extra"].pop("chat_state")
+                assert state in CHAT_STATES
+                message.addElement((NS_CHAT_STATES, state))
             return mess_data
 
-        post_xml_treatments.addCallback(treatment)
+        post_xml_treatments.addCallback(
+            lambda _: defer.ensureDeferred(treatment(mess_data))
+        )
         return True
 
-    def _is_muc(self, to_jid, profile):
+    def _is_muc(self, to_jid: JID, profile: str) -> bool:
         """Tell if that JID is a MUC or not
 
-        @param to_jid (JID): full or bare JID to check
-        @param profile (str): %(doc_profile)s
+        @param to_jid: full or bare JID to check
+        @param profile: %(doc_profile)s
         @return: bool
         """
         client = self.host.get_client(profile)
@@ -265,92 +419,114 @@ class XEP_0085(object):
             pass
         return False
 
-    def _check_activation(self, to_jid, forceEntityData, profile):
-        """
+    async def _check_activation(
+        self, client: SatXMPPEntity, to_jid: JID, forceEntityData: bool
+    ) -> bool:
+        """Check whether chat state notifications should be sent to a contact.
+
+        Activation follows a priority order:
+        1. User config
+        2. MUC
+        3. Disco cache (preferred)
+        4. Implicit discovery cache (fallback)
+
+        @param client: the client instance
         @param to_jid: the contact's full JID (or bare if you know it's a MUC)
-        @param forceEntityData: if set to True, a non-existing
-        entity data will be considered to be True (and initialized)
-        @param: current profile
-        @return: True if the notifications should be sent to this JID.
+        @param forceEntityData: if True, a missing implicit cache entry is
+            treated as True and initialized (used for initial discovery request)
+        @return: True if notifications should be sent to this JID.
         """
-        client = self.host.get_client(profile)
-        # check if the parameter is active
+        profile = client.profile
+
+        # XEP-0085 §5.2: user must be able to disable globally
         if not self.host.memory.param_get_a(PARAM_NAME, PARAM_KEY, profile_key=profile):
             return False
-        # check if notifications should be sent to this contact
-        if self._is_muc(to_jid, profile):
+
+        # XEP-0085 §5.5: MUC groupchat, no discovery required.
+        # MUC DMs (to_jid has resource) go through normal checks.
+        if self._is_muc(to_jid, profile) and not to_jid.resource:
             return True
-        # FIXME: this assertion crash when we want to send a message to an online bare jid
-        # assert to_jid.resource or not self.host.memory.is_entity_available(to_jid, profile) # must either have a resource, or talk to an offline contact
+
+        # XEP-0085 §4: explicit disco discovery (preferred over implicit)
+        disco_result = await self._check_disco_support(client, to_jid)
+        if disco_result is True:
+            # disco confirmed support, activate unconditionally
+            self.update_cache(to_jid, True, profile=profile)
+            return True
+        if disco_result is False:
+            # disco confirmed no support, deactivate
+            return False
+        # disco_result is None: no resources known, fall through to implicit discovery
+
+        # XEP-0085 §5.1: implicit discovery (fallback)
         try:
             return self.host.memory.get_entity_datum(client, to_jid, ENTITY_KEY)
         except (exceptions.UnknownEntityError, KeyError):
             if forceEntityData:
-                # enable it for the first time
+                # XEP-0085 §5.1 Rule 1: enable for initial discovery request
                 self.update_cache(to_jid, True, profile=profile)
                 return True
-        # wait for the first message before sending states
-        return False
+            # no discovery completed yet, wait for first message
+            return False
 
-    def _chat_state_init(self, to_jid, mess_type, profile):
+    def _chat_state_init(
+        self, client: SatXMPPEntity, to_jid: JID, mess_type: str
+    ) -> None:
+        """Data initialization for the chat state machine.
+
+        @param client: the client instance
+        @param to_jid: full JID for one2one, bare JID for MUC
+        @param mess_type: "chat" or "groupchat"
         """
-        Data initialization for the chat state machine.
+        chat_states_data = client.get_plugin_data(IMPORT_NAME, ChatStatesData)
+        machines = chat_states_data.machines
+        if to_jid not in machines:
+            machines[to_jid] = ChatStateMachine(
+                self.host, to_jid, mess_type, client.profile
+            )
 
-        @param to_jid (JID): full JID for one2one, bare JID for MUC
-        @param mess_type (str): "one2one" or "groupchat"
-        @param profile (str): %(doc_profile)s
+    def _chat_state_active(
+        self, client: SatXMPPEntity, to_jid: JID, mess_type: str
+    ) -> None:
+        """Launch the chat state machine on "active" state.
+
+        @param client: the client instance
+        @param to_jid: full JID for one2one, bare JID for MUC
+        @param mess_type: "chat" or "groupchat"
         """
-        if mess_type is None:
-            return
-        profile_map = self.map.setdefault(profile, {})
-        if to_jid not in profile_map:
-            machine = ChatStateMachine(self.host, to_jid, mess_type, profile)
-            self.map[profile][to_jid] = machine
+        self._chat_state_init(client, to_jid, mess_type)
+        chat_states_data = client.get_plugin_data(IMPORT_NAME, ChatStatesData)
+        chat_states_data.machines[to_jid]._onEvent("active")
 
-    def _chat_state_active(self, to_jid, mess_type, profile_key):
-        """
-        Launch the chat state machine on "active" state.
-
-        @param to_jid (JID): full JID for one2one, bare JID for MUC
-        @param mess_type (str): "one2one" or "groupchat"
-        @param profile (str): %(doc_profile)s
-        """
-        profile = self.host.memory.get_profile_name(profile_key)
-        if profile is None:
-            raise exceptions.ProfileUnknownError
-        self._chat_state_init(to_jid, mess_type, profile)
-        self.map[profile][to_jid]._onEvent("active")
-
-    def chat_state_composing(self, to_jid_s, profile_key):
+    async def chat_state_composing(self, to_jid_s: str, profile_key: str) -> None:
         """Move to the "composing" state when required.
 
-        Since this method is called from the front-end, it needs to check the
-        values of the parameter "Send chat state notifications" and the entity
-        data associated to the target JID.
+        Called from the frontend. Checks user config, disco support,
+        and implicit discovery state before sending.
 
-        @param to_jid_s (str): contact full JID as a string
-        @param profile_key (str): %(doc_profile_key)s
+        @param to_jid_s: contact full JID as a string
+        @param profile_key: %(doc_profile_key)s
         """
-        # TODO: try to optimize this method which is called often
         client = self.host.get_client(profile_key)
         to_jid = JID(to_jid_s)
-        if self._is_muc(to_jid, client.profile):
-            to_jid = to_jid.userhostJID()
-        elif not to_jid.resource:
-            to_jid.resource = self.host.memory.main_resource_get(client, to_jid)
-        if not self._check_activation(
-            to_jid, forceEntityData=False, profile=client.profile
-        ):
+        profile = client.profile
+
+        # XEP-0085 §4: fire disco check if not already done
+        await self._ensure_disco_check(client, to_jid)
+
+        if not await self._check_activation(client, to_jid, forceEntityData=False):
             return
-        try:
-            self.map[client.profile][to_jid]._onEvent("composing")
-        except (KeyError, AttributeError):
-            # no message has been sent/received since the notifications
-            # have been enabled, it's better to wait for a first one
-            pass
+
+        # Ensure state machine exists (may not be initialised if no message
+        # has been exchanged yet but disco confirmed support)
+        # groupchat (bare JID) vs MUC DM (full JID with resource)
+        is_groupchat = self._is_muc(to_jid, profile) and not to_jid.resource
+        self._chat_state_init(client, to_jid, "groupchat" if is_groupchat else "chat")
+        chat_states_data = client.get_plugin_data(IMPORT_NAME, ChatStatesData)
+        chat_states_data.machines[to_jid]._onEvent("composing")
 
 
-class ChatStateMachine(object):
+class ChatStateMachine:
     """
     This class represents a chat state, between one profile and
     one target contact. A timer is used to move from one state
@@ -360,7 +536,9 @@ class ChatStateMachine(object):
     frontend). Other states are automatically set with the timer.
     """
 
-    def __init__(self, host, to_jid, mess_type, profile):
+    def __init__(
+        self, host: "LiberviaBackend", to_jid: JID, mess_type: str, profile: str
+    ) -> None:
         """
         Initialization need to store the target, message type
         and a profile for sending later messages.
@@ -370,9 +548,9 @@ class ChatStateMachine(object):
         self.mess_type = mess_type
         self.profile = profile
         self.state = None
-        self.timer = None
+        self.timer: DelayedCall | None = None
 
-    def _onEvent(self, state):
+    def _onEvent(self, state: str) -> None:
         """
         Move to the specified state, eventually send the
         notification to the contact (the "active" state is
@@ -400,18 +578,21 @@ class ChatStateMachine(object):
                     "subject": {},
                     "extra": {},
                 }
-                client.generate_message_xml(mess_data)
-                mess_data["xml"].addElement((NS_CHAT_STATES, state))
-                client.send(mess_data["xml"])
+                client.generate_message_xml(cast(Any, mess_data))
+                message_elt = cast(domish.Element, mess_data["xml"])
+                message_elt.addElement((NS_CHAT_STATES, state))
+                client.send(message_elt)
 
         self.state = state
-        try:
-            self.timer.cancel()
-        except (internet_error.AlreadyCalled, AttributeError):
-            pass
+        if self.timer is not None:
+            try:
+                self.timer.cancel()
+            except internet_error.AlreadyCalled:
+                pass
 
         if transition["next_state"] and transition["delay"] > 0:
-            self.timer = reactor.callLater(
+            reactor_time = cast(IReactorTime, reactor)
+            self.timer = reactor_time.callLater(
                 transition["delay"], self._onEvent, transition["next_state"]
             )
 
@@ -419,13 +600,15 @@ class ChatStateMachine(object):
 @implementer(itx_xmpp.IDisco)
 class XEP_0085_handler(XMPPHandler):
 
-    def __init__(self, plugin_parent, profile):
+    def __init__(self, plugin_parent: "XEP_0085", profile: str) -> None:
         self.plugin_parent = plugin_parent
         self.host = plugin_parent.host
         self.profile = profile
 
-    def getDiscoInfo(self, requestor, target, nodeIdentifier=""):
+    def getDiscoInfo(self, requestor: JID, target: JID, nodeIdentifier: str = "") -> list:
         return [disco.DiscoFeature(NS_CHAT_STATES)]
 
-    def getDiscoItems(self, requestor, target, nodeIdentifier=""):
+    def getDiscoItems(
+        self, requestor: JID, target: JID, nodeIdentifier: str = ""
+    ) -> list:
         return []