Restructure config and registration around ChattoConfiguration

Config handling becomes declarative: a ConfigField descriptor resolves
env var -> PlatformConfig.extra -> default per instance, and
ChattoConfiguration declares every setting as one line. The Hermes
module hooks (hermes_validate_config_fn, hermes_check_fn,
hermes_is_connected, hermes_apply_yaml_config_fn) and register() move
here from the adapter, so __init__.py re-exports register from
platform_config.

__init__.py

@@ -1,3 +1,4 @@
-from .adapter import register
+from .platform_config import register
+
+__all__ = ["register"]
 
-__all__ = ["register"]

platform_config.py

@@ -4,67 +4,421 @@ Chatto Platform Config
 
 """
 from dataclasses import dataclass
+import logging
 import os
-from typing import Optional
+from typing import Any, Dict, Optional, cast
 
+from gateway.config import PlatformConfig
 import utils
 
+from vendor.chattolib.client import ChattoClient
 
+from .adapter import ChattoAdapter, hermes_standalone_sender_fn
+
+logger = logging.getLogger(__name__)
+
+# --------------------------------------------------------------------------- #
+# Constants
+# --------------------------------------------------------------------------- #
+
+class ChattoConstants:
+    """
+    Chatto Platform Constants
+    """
+
+    def __init__(self):
+        pass
+
+    PLATFORM_ID: str = "chatto"
+    PLATFORM_NAME: str = "chatto-platform"
+    PLATFORM_LABEL: str = "Chatto"
+
+    INSTALL_HINT = "Requires a Chatto server. See https://docs.chatto.run"
+
+    EXTRA_ENV_MAPPING = {
+        "base_url": "CHATTO_BASE_URL",
+        "home_channel": "CHATTO_HOME_CHANNEL",
+        "require_mention": "CHATTO_REQUIRE_MENTION",
+        "free_response_channels": "CHATTO_FREE_RESPONSE_CHANNELS",
+        "auto_thread": "CHATTO_AUTO_THREAD",
+        "allow_all_users": "CHATTO_ALLOW_ALL_USERS",
+        "allowed_users": "CHATTO_ALLOWED_USERS",
+    }
+
+    MAX_MESSAGE_LENGTH = 10000
+    SEEN_CAP = 500
+
+    # WebSocket / realtime protocol
+    WS_PATH = "/api/realtime"
+    WS_AUTH_TIMEOUT = 20.0
+    WS_MAX_MESSAGE_BYTES = 4_000_000
+    WS_PING_INTERVAL = 30.0
+    WS_RECONNECT_INITIAL_BACKOFF = 1.0
+    WS_RECONNECT_MAX_BACKOFF = 30.0
+
+    # HTTP timeout used for outbound URL fetches
+    HTTP_TIMEOUT = 30
+
+    # Emoji shortcode mapping (Chatto uses shortcode names, not unicode emoji)
+
+    # Emoji shortcode mapping (Chatto uses shortcode names, not unicode emoji)
+    EMOJI_TO_SHORTCODE: Dict[str, str] = {
+        "👍": "thumbsup",
+        "👎": "thumbsdown",
+        "❤️": "heart",
+        "❤": "heart",
+        "✅": "white_check_mark",
+        "❌": "x",
+        "👀": "eyes",
+        "🎉": "tada",
+        "😂": "joy",
+        "🚀": "rocket",
+        "🔥": "fire",
+        "💯": "100",
+        "🤔": "thinking",
+        "👏": "clap",
+        "🙏": "pray",
+        "😅": "sweat_smile",
+        "😴": "sleeping",
+        "⏳": "hourglass",
+    }
+
+    # Chunk size for asset uploads (256 KB)
+    UPLOAD_CHUNK_SIZE = 256 * 1024
+
+
+def _get_env_or_extra_str_opt(env_var: str, extra_val: Optional[str], default: Optional[str] = None) -> Optional[str]:
+    """Get a value from environment variable or extra config."""
+    env_value = os.getenv(env_var)
+    if env_value is not None:
+        return env_value.strip()
+
+    if extra_val is not None:
+        return extra_val.strip()
+
+    if default is not None:
+        return default.strip() 
+    return None
+
+def _get_env_or_extra_str(env_var: str, extra_val: Optional[str], default: Optional[str] = None) -> str:
+    """Get a value from environment variable or extra config."""
+    my_string = _get_env_or_extra_str_opt(env_var, extra_val, default)
+    if my_string:
+        return my_string
+    return ""
+
+
+def _get_env_or_extra_truthy(env_var: str, extra_val: Optional[str], default: bool = False) -> bool:
+    """Get a boolean value from environment variable or extra config."""
+    return utils.is_truthy_value(_get_env_or_extra_str(env_var, extra_val, "False"), default)
+
+
+def _split_str_to_list(mystring: str) -> list:
+    return list(c.strip() for c in mystring.split(","))
+
+
+def _get_env_or_extra_list(env_var: str, extra_val: Optional[set]) -> list[str]:
+    """Get a list of values from environment variable or extra config."""
+    env_value = os.getenv(env_var)
+    if env_value is not None:
+        return _split_str_to_list(env_value)
+
+    if extra_val is not None:
+        if isinstance(extra_val, list):
+            return list(
+                c.strip() for c in extra_val
+            )
+        if isinstance(extra_val, str):
+            return _split_str_to_list(extra_val)
+
+    return []
+
+from typing import TypeVar, Generic, Any
+
+# 1. Define a generic Type Variable
+T = TypeVar('T')
+
+# 2. Inherit from Generic[T]
+class ConfigField(Generic[T]):
+    """Repräsentiert ein einzelnes Konfigurationsfeld mit IDE-Support."""
+# 3. Type 'value' as T (or T | None since it starts as None)
+    value: T
+    field_name: str = ""
+    env_name: str = ""
+
+    # 4. Hint that the init argument should match type T
+    # We add `| Any` as a fallback because complex types like `list[str]` 
+    # can sometimes confuse older type checkers when used with `type[T]`
+    def __init__(self, type_: type[T] | Any):
+        self._type = type_
+
+    # 5. Ensure the IDE knows only type T can be assigned
+    def __set__(self, instance: Any, value: T) -> None:
+        self.value = value
+        self._type = type(value)
+
+    def __str__(self) -> str:
+        return str(self.value)
+
+    def __set_name__(self, owner: Any, name: str) -> None:
+        temp: str = str(name).lower().replace("chatto_", "", 1)
+        self.field_name = temp
+        self.env_name = f"CHATTO_{temp.upper()}"
 
 
 @dataclass
-class ChattoConfig:
+class ChattoConfiguration:
     """
     Chatto Platform Config
 
     Some constants and getting from environment/config.yaml
     """
+    base_url = ConfigField(str)
+    token = ConfigField(str | None)
+    login = ConfigField(str)
+    password = ConfigField(str)
+    channels = ConfigField(str)
+    channels_list = ConfigField(list[str])
+    home_channel = ConfigField(str)
+    allowed_users = ConfigField(str)
+    require_mention = ConfigField(bool)
+    free_response_channels_list = ConfigField(list[str])
+    auto_thread = ConfigField(bool)
+    allow_all_users = ConfigField(bool)
+    reactions = ConfigField(bool)
 
-    _token: Optional[str] = None
-    _PLATFORM: str = "CHATTO"
-
-    def __init__(self, extra: dict = {}):
-
-        self._base_url = os.getenv("CHATTO_BASE_URL", "").strip() # todo: add config.yaml support
-        self._token = os.getenv("CHATTO_TOKEN", "").strip() # todo: add config.yaml support 
-        self._login = os.getenv("CHATTO_LOGIN", "").strip() # todo: add config.yaml support
-        self._password = os.getenv("CHATTO_PASSWORD", "").strip() # todo: add config.yaml support
+    def __init__(self, pconfig: PlatformConfig):
+        """PlatformConfig from Hermes provides our own configuration within the "extra" 
+        object. But here, we allow overriding via environment variables again.
 
-        self._raw_channels = os.getenv("CHATTO_CHANNELS", "").strip() # todo: add config.yaml support
+        We take our Configuration from this typed Class, because it is easier access than using extra.get["base_url"].
+        """
+        self.base_url.value = _get_env_or_extra_str(self.base_url.env_name,
+            pconfig.extra.get(self.base_url.field_name), ChattoClient.DEFAULT_BASE_URL)
 
-        if self._raw_channels:
-            self._channel_ids = [c.strip() for c in self._raw_channels.split(",") if c.strip()]
-        elif isinstance(extra.get("channels"), list):
-            self._channel_ids = [str(c) for c in extra["channels"]]
-        else:
-            self._channel_ids = []
+        self.token.value = _get_env_or_extra_str_opt(self.token.env_name, pconfig.extra.get(self.token.field_name))
+        self.login.value = _get_env_or_extra_str(self.login.env_name, pconfig.extra.get(self.login.field_name))
+        self.password.value = _get_env_or_extra_str(self.password.env_name, pconfig.extra.get(self.password.field_name))
+        
+        self.channels.value = str(_get_env_or_extra_list(self.channels.env_name, pconfig.extra.get(self.channels.field_name)))
+        self.channels_list.value = _get_env_or_extra_list(self.channels.env_name, pconfig.extra.get(self.channels.field_name))
 
+        self.home_channel.value = _get_env_or_extra_str(self.home_channel.env_name, pconfig.extra.get(self.home_channel.field_name))
 
-        self._home_channel = (
-            os.getenv("CHATTO_HOME_CHANNEL", "").strip()
-            or str(extra.get("home_channel", "")).strip()
-        )
+        self.allowed_users.value = _get_env_or_extra_str(self.allowed_users.env_name, pconfig.extra.get(self.allowed_users.field_name))
 
-        self._require_mention = os.getenv("CHATTO_REQUIRE_MENTION", "").strip().lower()
-        if self._require_mention:
-            self._require_mention = self._require_mention in ("true", "1", "yes")
-        else:
-            self._require_mention = bool(extra.get("require_mention", True))
+        self.require_mention.value = _get_env_or_extra_truthy(self.require_mention.env_name, pconfig.extra.get(self.require_mention.field_name))
 
         # free_response_channels: room IDs where the bot responds without being mentioned via "@botname"
-        _free_response = os.getenv("CHATTO_FREE_RESPONSE_CHANNELS", "").strip()
-        if _free_response:
-            self._free_response_channels = set(c.strip() for c in _free_response.split(",") if c.strip())
-        else:
-            self._free_response_channels = set(
-                str(c) for c in extra.get("free_response_channels", []) if str(c).strip()
-            )
+        self.free_response_channels_list.value = _get_env_or_extra_list(self.free_response_channels_list.env_name,
+                                                                   pconfig.extra.get(self.free_response_channels_list.field_name))
 
         # Auto-thread: by default, Chatto creates a thread for replies to room
         # messages (not DMs, not already in a thread). This keeps conversations
         # organized in the room. Can be disabled via extra.auto_thread=false.
-        self._auto_thread = utils.is_truthy_value(os.getenv("CHATTO_AUTO_THREAD", "").strip().lower()) # todo: add config.yaml support
+        self.auto_thread.value = _get_env_or_extra_truthy(self.auto_thread.env_name, pconfig.extra.get(self.auto_thread.field_name))
+
+        self.allow_all_users.value = _get_env_or_extra_truthy(self.allow_all_users.env_name, pconfig.extra.get(self.allow_all_users.field_name))
+        self.reactions.value = _get_env_or_extra_truthy(self.reactions.env_name, pconfig.extra.get(self.reactions.field_name))
+
+
+def hermes_validate_config_fn(config: PlatformConfig) -> bool:
+    """"
+    Function name should be the same as register argument name with "hermes_" prefix, so we
+    know that it is needed for plugin register(). Do not change signature.
+    - config
+    Check whether Chatto Plugin is configured."""
+
+    chatto_config = ChattoConfiguration(pconfig=config)
+
+    if chatto_config.base_url.value and (chatto_config.token.value or (chatto_config.login.value and chatto_config.password.value)):
+        return True
+    return False
+
+
+def hermes_check_fn() -> bool:
+    """Check if Chatto is configured and dependencies are available.
+    Add real logic?! Or just .. there are no dependencies.. always return true. Really. Docs suck."""
+    try:
+        from vendor.chattolib.client import ChattoClient
+        return True
+    except ImportError:
+        return False
+    return True
+
+
+# ---------------------------------------------------------------------------
+# is_connected probe
+# ---------------------------------------------------------------------------
+
+def hermes_is_connected(config: PlatformConfig) -> bool:
+    """Check whether Chatto Plugin is connected. But where to? To the Hermes Agent? To Chatto Server?
+    The Hermes Agent plugin docs suck and it seems there are many functions to do the same."""
+    return bool(hermes_validate_config_fn(config) and config.enabled)
+
+
+
+# ---------------------------------------------------------------------------
+# YAML → env config bridge
+# ---------------------------------------------------------------------------
+
+@DeprecationWarning
+def hermes_apply_yaml_config_fn(yaml_dict: dict, platform_dict: dict) -> Optional[dict]:
+    """Translate config.yaml chatto.extra keys into CHATTO_* env vars.
+    I don't actually get why Hermes wants us to modify OS environment variables.
+    Bad behavior in my book.
+    
+    Also .. I don't think we need this"""
+
+    if not isinstance(platform_dict, dict):
+        platform_dict = {}
+    extra = platform_dict.get("extra", {}) or {}
+    if not isinstance(extra, dict):
+        extra = {}
+
+    for yaml_key, env_key in ChattoConstants.EXTRA_ENV_MAPPING.items():
+        val = extra.get(yaml_key)
+        if val is not None and not os.getenv(env_key):
+            if isinstance(val, bool):
+                env_val = str(val).lower()
+            elif isinstance(val, list):
+                env_val = ",".join(str(v) for v in val)
+            else:
+                env_val = str(val)
+            os.environ[env_key] = env_val
+
+    channels = extra.get(ChattoConfiguration.channels.field_name)
+    if isinstance(channels, list) and not os.getenv(ChattoConfiguration.channels.env_name):
+        os.environ[ChattoConfiguration.channels.env_name] = ",".join(str(c) for c in channels)
+
+    allowed = extra.get(ChattoConfiguration.allowed_users.field_name)
+    if isinstance(allowed, list) and not os.getenv(ChattoConfiguration.allowed_users.env_name):
+        os.environ[ChattoConfiguration.allowed_users.env_name] = ",".join(str(u) for u in allowed)
+
+    if ChattoConfiguration.allow_all_users.field_name in extra and not os.getenv(ChattoConfiguration.allow_all_users.env_name):
+        os.environ[ChattoConfiguration.allow_all_users.env_name] = str(extra[ChattoConfiguration.allow_all_users.field_name]).lower()
+
+    return None
+
+
+def hermes_setup_fn() -> None:
+    """Interactive setup wizard for Chatto. Is called by and only works in Hermes CLI context.
+    Function name should be the same as register argument name with "hermes_" prefix, so we
+    know that it is needed for plugin register().
+    """
+    from hermes_cli.setup import (
+        prompt,
+        prompt_yes_no,
+        save_env_value,
+        get_env_value,
+        print_header,
+        print_info,
+        print_warning,
+        print_success,
+    )
+
+    url = prompt(
+        "Chatto server URL (e.g. https://chat.example.com) or leave blank for default ChattoHQ on chat.chatto.run:")
+    if url:
+        save_env_value(ChattoConfiguration.base_url.env_name, url)
+
+    login = prompt("Chatto login (username):")
+    if login:
+        save_env_value(ChattoConfiguration.login.env_name, login)
+
+    password = prompt("Chatto password:", password=True)
+    if password:
+        save_env_value(ChattoConfiguration.password.env_name, password)
+
+    channels = prompt("Room IDs to watch (comma-separated, or empty for all):")
+    if channels:
+        save_env_value(ChattoConfiguration.channels.env_name, channels)
+
+    home = prompt("Home room ID for notifications (or empty):")
+    if home:
+        save_env_value(ChattoConfiguration.home_channel.env_name, home)
+
+    allow_all = prompt_yes_no("Allow all users to talk? (true/false):")
+    if allow_all:
+        save_env_value(ChattoConfiguration.allow_all_users.env_name, str(allow_all))
+    print_success("\n✓ Chatto configured. Restart the gateway to activate.")
+
+
+def hermes_env_enablement_fn() -> Optional[dict]:
+    """Seed PlatformConfig.extra from env vars.
+
+    Returns a dict compatible with the PlatformConfig merge hook (or None
+    when no env-provided values are present).
+
+    Called by the platform registry during load_gateway_config().
+    Return None when the platform isn't minimally configured — the
+    caller then skips auto-enabling. Return a dict to seed extras.
+
+    The special 'home_channel' key is extracted and becomes a proper
+    HomeChannel dataclass on the PlatformConfig; every other key is
+    merged into PlatformConfig.extra.
+
+    Function name should be the same as register argument name with "hermes_" prefix, so we
+    know that it is needed for plugin register().
+    """
+
+    def _add_env_to_seed(seed: dict, our_key: str) -> dict:
+        env_value = os.getenv(our_key.upper())
+        if env_value:
+            seed[our_key.lower()] = env_value
+        return seed
+    
+    seed = {}
+    seed["base_url"] = (os.getenv(ChattoConfiguration.base_url.env_name) or ChattoClient.DEFAULT_BASE_URL).strip()
+
+    seed = _add_env_to_seed(seed, ChattoConfiguration.token.env_name)
+    seed = _add_env_to_seed(seed, ChattoConfiguration.login.env_name)
+    seed = _add_env_to_seed(seed, ChattoConfiguration.password.env_name)
+
+    seed = _add_env_to_seed(seed, ChattoConfiguration.channels.env_name)
+    seed = _add_env_to_seed(seed, ChattoConfiguration.home_channel.env_name)
+    seed = _add_env_to_seed(seed, ChattoConfiguration.require_mention.env_name)
+    seed = _add_env_to_seed(seed, ChattoConfiguration.free_response_channels_list.env_name)
+    seed = _add_env_to_seed(seed, ChattoConfiguration.auto_thread.env_name)
+
+    return seed
+
 
+# ---------------------------------------------------------------------------
+# Plugin registration entry point
+# ---------------------------------------------------------------------------
 
+def hermes_adapter_factory(config: PlatformConfig):
+    """Factory wrapper that constructs ChattoAdapter from a PlatformConfig."""
+    return ChattoAdapter(config)
 
 
+def register(ctx) -> None:
+    """Plugin entry point — called by the Hermes plugin system."""
+    logger.error("Registering Chatto Plugin")
+    ctx.register_platform(
+        name=ChattoConstants.PLATFORM_NAME,
+        label=ChattoConstants.PLATFORM_LABEL,
+        adapter_factory=hermes_adapter_factory,
+        check_fn=hermes_check_fn,
+        validate_config=hermes_validate_config_fn,
+        is_connected=hermes_is_connected,
+        install_hint=ChattoConstants.INSTALL_HINT,
+        env_enablement_fn=hermes_env_enablement_fn,
+        setup_fn=hermes_setup_fn,
+        apply_yaml_config_fn=hermes_apply_yaml_config_fn,
+        cron_deliver_env_var=ChattoConfiguration.home_channel.env_name,
+        standalone_sender_fn=hermes_standalone_sender_fn,
+        allowed_users_env=ChattoConfiguration.allowed_users.env_name,
+        allow_all_env=ChattoConfiguration.allow_all_users.env_name,
+        max_message_length=ChattoConstants.MAX_MESSAGE_LENGTH,
+        emoji="💬",
+        allow_update_command=True,
+        pii_safe=False,
+        platform_hint=(
+            "You are chatting in Chatto (a self-hosted or cloud-hosted team or community chat server). "
+            "Markdown is supported. Users _may_ address you by @-mentioning your name. If configured, " \
+            "you also react without a @-mention. Direct messages reach you without a mention."
+            "Keep responses conversational."
+        ),
+    )

plugin.yaml

@@ -4,16 +4,16 @@ kind: platform
 version: 1.0.0
 description: >
   Chatto gateway adapter for Hermes Agent.
-  Connects to a Chatto server (self-hosted team chat) and relays messages
-  between rooms/DMs and the Hermes agent.  Uses the Chatto ConnectRPC API
+  Connects to a Chatto server (cloud-hosted or self-hosted team chat) and relays messages
+  between rooms/DMs and the Hermes agent. Uses the Chatto ConnectRPC API
   (JSON over HTTP) for outbound messages and the Chatto realtime WebSocket
   protocol (binary protobuf) for inbound events.  No Python packages
-  required beyond websockets — pure stdlib protobuf codec.
-author: Nous Research
+  required beyond websockets.
+author: Chatto community guys
 requires_env:
-  - name: CHATTO_URL
-    description: "Base URL of the Chatto server (e.g. https://chat.example.com)"
-    prompt: "Chatto server URL"
+  - name: CHATTO_BASE_URL
+    description: "Base URL of the Chatto server (e.g. https://chat.example.com) leave blank for ChattoHQ (https://chat.chatto.run)"
+    prompt: "Chatto server Base URL"
     password: false
   - name: CHATTO_LOGIN
     description: "Chatto login (username)"

test_platform_config.py

@@ -0,0 +1,64 @@
+import os
+import sys
+from typing import Optional
+
+import pytest
+
+PLUGIN_ROOT = os.path.abspath(os.path.dirname(__file__))
+sys.path.insert(0, PLUGIN_ROOT)
+sys.path.insert(0, "/opt/hermes")
+sys.path.insert(0, "/root/.hermes/plugins/platforms/chatto")
+
+from platform_config import (
+    _get_env_or_extra_str,
+    _get_env_or_extra_truthy,
+    _split_str_to_list,
+    _get_env_or_extra_list,
+)
+
+
+class TestPlatformConfigHelpers:
+    """Test Chatto configuration helper functions."""
+
+    def test_get_env_or_extra_str_prefers_env(self, monkeypatch):
+        monkeypatch.setenv("CHATTO_TEST", " env-value ")
+        assert _get_env_or_extra_str("CHATTO_TEST", "extra-value") == "env-value"
+
+    def test_get_env_or_extra_str_uses_extra_when_no_env(self, monkeypatch):
+        monkeypatch.delenv("CHATTO_TEST", raising=False)
+        assert _get_env_or_extra_str("CHATTO_TEST", " extra-value ") == "extra-value"
+
+    def test_get_env_or_extra_str_returns_none_when_missing(self, monkeypatch):
+        monkeypatch.delenv("CHATTO_TEST", raising=False)
+        assert _get_env_or_extra_str("CHATTO_TEST", None) is None
+
+    def test_get_env_or_extra_truthy_parses_truthy_values(self, monkeypatch):
+        monkeypatch.setenv("CHATTO_TEST", " yes ")
+        assert _get_env_or_extra_truthy("CHATTO_TEST", None) is True
+
+    def test_get_env_or_extra_truthy_uses_extra_when_no_env(self, monkeypatch):
+        monkeypatch.delenv("CHATTO_TEST", raising=False)
+        assert _get_env_or_extra_truthy("CHATTO_TEST", " true ") is True
+
+    def test_get_env_or_extra_truthy_uses_default_false(self, monkeypatch):
+        monkeypatch.delenv("CHATTO_TEST", raising=False)
+        assert _get_env_or_extra_truthy("CHATTO_TEST", "false") is False
+
+    def test_split_str_to_list_handles_comma_separated_values(self):
+        assert _split_str_to_list("a,b, c ,d") == ["a", "b", "c", "d"]
+
+    def test_get_env_or_extra_list_prefers_env(self, monkeypatch):
+        monkeypatch.setenv("CHATTO_TEST", " one,two , three")
+        assert _get_env_or_extra_list("CHATTO_TEST", ["ignored", "list"]) == ["one", "two", "three"]
+
+    def test_get_env_or_extra_list_uses_extra_list(self, monkeypatch):
+        monkeypatch.delenv("CHATTO_TEST", raising=False)
+        assert _get_env_or_extra_list("CHATTO_TEST", ["a", " b "]) == ["a", "b"]
+
+    def test_get_env_or_extra_list_uses_extra_string(self, monkeypatch):
+        monkeypatch.delenv("CHATTO_TEST", raising=False)
+        assert _get_env_or_extra_list("CHATTO_TEST", "x,y") == ["x", "y"]
+
+    def test_get_env_or_extra_list_returns_empty_for_none(self, monkeypatch):
+        monkeypatch.delenv("CHATTO_TEST", raising=False)
+        assert _get_env_or_extra_list("CHATTO_TEST", None) == []