chatter

.python-version

@@ -0,0 +1 @@
+3.14

colony/cli.py

@@ -1,9 +1,12 @@
 import argparse
+import datetime
 import json
 import sys
 import time
 
 from colony.core.network.requests import CommandRequest
+from colony.core.roles.chatter.role import Chatter
+from colony.core.storages.chat import ChatStorage
 from colony.core.roles.unit.role import Unit
 from colony.settings import configuration, setting
 from colony.transport.whisperer import Whisperer
@@ -16,11 +19,11 @@ CLEAR_SCREEN = "\033[2J"
 CLEAR_LINE = "\033[K"
 CLEAR_BELOW = "\033[J"
 
-def call(host, port, command, *args, timeout=DEFAULT_TIMEOUT, cert_file=None, keyfile=None):
-    payload = CommandRequest.form(command, *args).collapse()
+def call(host, port, command, *args, sender=None, timeout=DEFAULT_TIMEOUT, cert_file=None, keyfile=None):
+    payload = CommandRequest.form(command, *args).collapse(sender=sender)
     resp = Whisperer(host, port, cert_file, keyfile).send(payload, timeout=timeout)
 
-    if not isinstance(resp, dict) or resp.get("status") != "ok" or not isinstance(resp.get("result"), dict):
+    if not isinstance(resp, dict) or resp.get("status") != "ok" or "result" not in resp:
         reason = resp.get("error") if isinstance(resp, dict) else resp
         raise RuntimeError(f"the node answered with an error: {reason}")
     return resp["result"]
@@ -34,6 +37,61 @@ def request_role(host, port, role, timeout=DEFAULT_TIMEOUT, cert_file=None, keyf
     return call(host, port, Unit.COMMANDS.REQUEST_ROLE, role, timeout=timeout, cert_file=cert_file, keyfile=keyfile)
 
 
+def chat_history(peer_id):
+    return ChatStorage().messages(peer_id)
+
+
+def send_chat(own, target, text, timeout=DEFAULT_TIMEOUT):
+    call(target["ip"], target["port"], Chatter.COMMANDS.RECEIVE, text, sender=member_id(own), timeout=timeout)
+    ChatStorage().append(member_id(target), "out", text)
+
+
+def member_id(member):
+    return f"{member['hostname']}@{member['ip']}"
+
+
+def resolve_chat_target(payload, name):
+    members = payload["members"]
+    own = next(member for member in members if member["self"])
+    if "Chatter" not in own["roles"]:
+        raise LookupError(f"{own['hostname']} is not running Chatter. Start it with: colonyctl role add Chatter")
+
+    matches = [member for member in members if name in (address(member), member_id(member))]
+    if not matches:
+        matches = [member for member in members if member["hostname"].lower() == name.lower()]
+    chatters = [member["hostname"] for member in members if "Chatter" in member["roles"] and not member["self"]]
+    if not matches:
+        raise LookupError(f"unknown node {name!r}. Nodes running Chatter: {', '.join(chatters) or 'none'}")
+    if len(matches) > 1:
+        raise LookupError(f"{name!r} matches more than one node, use an address: {', '.join(address(member) for member in matches)}")
+
+    target = matches[0]
+    if target["self"]:
+        raise LookupError("that is the node you are connected to, pick another node")
+    if "Chatter" not in target["roles"]:
+        raise LookupError(f"{target['hostname']} is not running Chatter. Nodes running it: {', '.join(chatters) or 'none'}")
+    return target
+
+
+def format_chat_time(at):
+    try:
+        moment = datetime.datetime.fromisoformat(at).astimezone()
+    except (TypeError, ValueError):
+        return "?"
+    if moment.date() == datetime.date.today():
+        return moment.strftime("%H:%M:%S")
+    return moment.strftime("%Y-%m-%d %H:%M")
+
+
+def render_chat(messages, own_label, peer_label):
+    if not messages:
+        return f"No messages with {peer_label} yet."
+    return "\n".join(
+        f"{format_chat_time(message.get('at'))}  {own_label if message.get('direction') == 'out' else peer_label}: {message.get('text', '')}"
+        for message in messages
+    )
+
+
 def resolve_on_demand_role(payload, name):
     roles = payload["roles"]
     match = next((role for role in roles if role["name"].lower() == name.lower()), None)
@@ -221,6 +279,13 @@ def build_parser():
     add.add_argument("name", help="role name, e.g. WebView (case does not matter)")
 
     commands.add_parser("webview", parents=[connection], help="ask the node to run the web view (same as: role add WebView)")
+
+    chat = commands.add_parser("chat", parents=[connection], help="show the chat with a node (live in a terminal), or send it a message; the history is read from this machine's storages/chat")
+    chat.add_argument("node", help="hostname, ip:port or hostname@ip of the node to chat with")
+    chat.add_argument("text", nargs="?", help="message to send; without it the history is shown")
+    chat.add_argument("--interval", type=float, default=DEFAULT_INTERVAL, help="seconds between refreshes")
+    chat.add_argument("--raw", action="store_true", help="print the history once and exit instead of refreshing")
+    chat.add_argument("--json", action="store_true", help="print the history as JSON once and exit")
     return parser
 
 
@@ -245,6 +310,51 @@ def watch(host, port, args, out):
         return 0
 
 
+def watch_chat(args, port, target, own, out):
+    peer_label, own_label = target["hostname"], f"{own['hostname']} (you)"
+    out.write(CLEAR_SCREEN)
+    try:
+        while True:
+            try:
+                text = render_chat(chat_history(member_id(target)), own_label, peer_label)
+            except (OSError, ValueError, RuntimeError) as e:
+                text = f"colonyctl: cannot read the chat: {e}\nretrying..."
+            footer = f"Chat with {peer_label}. Every {args.interval:g}s. Updated {time.strftime('%H:%M:%S')}. Ctrl-C to quit."
+            footer += f'\nSend with: colonyctl chat {target["hostname"]} "message"'
+            draw(out, f"{text}\n\n{footer}")
+            time.sleep(args.interval)
+    except KeyboardInterrupt:
+        out.write("\n")
+        return 0
+
+
+def chat_command(args, port, out):
+    try:
+        payload = fetch_cluster(args.host, port, timeout=args.timeout)
+        target = resolve_chat_target(payload, args.node)
+        own = next(member for member in payload["members"] if member["self"])
+        if args.text is not None:
+            send_chat(own, target, args.text, timeout=args.timeout)
+        elif not args.raw and not args.json and out.isatty():
+            return watch_chat(args, port, target, own, out)
+        else:
+            history = chat_history(member_id(target))
+    except LookupError as e:
+        print(f"colonyctl: {e}", file=sys.stderr)
+        return 1
+    except (OSError, ValueError, RuntimeError) as e:
+        print(f"colonyctl: cannot reach {args.host}:{port}: {e}", file=sys.stderr)
+        return 1
+
+    if args.text is not None:
+        print(f"Sent to {target['hostname']}", file=out)
+    elif args.json:
+        print(json.dumps(history, indent=2), file=out)
+    else:
+        print(render_chat(history, f"{own['hostname']} (you)", target["hostname"]), file=out)
+    return 0
+
+
 def add_role(args, port, name, out):
     try:
         payload = fetch_cluster(args.host, port, timeout=args.timeout)
@@ -277,6 +387,8 @@ def main(argv=None, out=None):
         return add_role(args, port, args.name, out)
     if args.command == "webview":
         return add_role(args, port, "WebView", out)
+    if args.command == "chat":
+        return chat_command(args, port, out)
 
     if not args.raw and not args.json and out.isatty():
         return watch(args.host, port, args, out)

colony/core/context.py

@@ -11,15 +11,9 @@ class RoleContext:
         return self._network.execute(command, *args, **kwargs)
 
     def execute_on(self, target, command, *args, **kwargs):
-        """Run a command on one particular node, going through a gateway if it cannot be dialled.
-
-        `target` is a Node or a MemberRecord, anything with `ip` and `port`, for example an item of
-        `nodes_with_role(...)`. `execute` picks the node for you; this one does not.
-        """
         return self._network.send_command(target.ip, target.port, command, *args, **kwargs)
 
     def execute_to_unregistered_node(self, ip, port, command, *args, **kwargs):
-        """The same as `execute_on` for a caller that only has an address."""
         return self._network.send_command(ip, port, command, *args, **kwargs)
 
     def send_direct(self, ip, port, payload, timeout=None, connect_timeout=None):
@@ -37,6 +31,9 @@ class RoleContext:
     def peers(self):
         return self._network.get_all_nodes()
 
+    def node_by_id(self, node_id):
+        return self._network.get_node_by_id(node_id)
+
     def nodes_with_role(self, role):
         return self._network.get_nodes_of_role(role if isinstance(role, str) else role.__name__)
 

colony/core/dispatcher.py

@@ -20,7 +20,11 @@ def parse_command(payload):
     if not isinstance(args, (list, tuple)) or not isinstance(kwargs, dict):
         raise ValueError("Command args must be a list and kwargs a dict")
 
-    return role_name, command_name, tuple(args), kwargs
+    caller_id = command.get("from")
+    if not isinstance(caller_id, str):
+        caller_id = None
+
+    return role_name, command_name, tuple(args), kwargs, caller_id
 
 
 class Dispatcher:
@@ -31,7 +35,7 @@ class Dispatcher:
         self.role_registry = role_registry
 
     def dispatch(self, payload):
-        role_name, command_name, args, kwargs = parse_command(payload)
+        role_name, command_name, args, kwargs, caller_id = parse_command(payload)
         self.logger.debug(f"Dispatching payload for command {role_name}.{command_name}")
 
         role_entry = self.role_registry.get_role_class_by_name(role_name)
@@ -44,6 +48,6 @@ class Dispatcher:
         node = self.network.get_self()
 
         self.logger.info(f"Executing {role_name}.{command_name} on local node {node.id}")
-        result = node.execute_local(command, *args, **kwargs)
+        result = node.execute_local(command, *args, caller_id=caller_id, **kwargs)
         self.logger.debug(f"Passing the results from {role_name}.{command_name}")
         return result

colony/core/network/core.py

@@ -8,7 +8,7 @@ from colony.core.network.relay import GATEWAY_ROLE, RelayError, build_forward, r
 from colony.core.network.requests import CommandRequest
 from colony.core.logger.logging import get_logger
 from colony.settings import configuration, setting
-from colony.transport.whisperer import ConnectFailed, Whisperer
+from colony.transport.whisperer import Whisperer, ConnectFailed
 from colony.utils.network import get_local_ip, get_hostname
 
 
@@ -56,6 +56,9 @@ class Network:
     def get_all_nodes(self):
         return self.nodes
 
+    def get_node_by_id(self, node_id: str) -> Optional[Node]:
+        return next((node for node in self.nodes if node.id == node_id), None)
+
     def get_nodes_of_role(self, role_name: str) -> List[Node]:
         return [node for node in self.nodes if role_name in node.roles]
 
@@ -125,7 +128,7 @@ class Network:
         return None
 
     def send_command(self, ip, port, command, *args, timeout=None, **kwargs):
-        payload = CommandRequest.form(command, *args, **kwargs).collapse()
+        payload = CommandRequest.form(command, *args, **kwargs).collapse(sender=self.self_node.id)
         key = (ip, port)
         now = self.membership.now()
         gateways = self._gateways(key, now)
@@ -184,14 +187,6 @@ class Network:
         self._direct_ok[key] = self.membership.now()
         return response
 
-    @staticmethod
-    def execute_to_unregistered_node(ip, port, command: Enum, *args, timeout=None, **kwargs):
-        request = CommandRequest.form(command, *args, **kwargs)
-
-        options = {} if timeout is None else {"timeout": timeout}
-        return Whisperer(ip, port).send(request.collapse(), **options)
-
-
     def load_network_from_snapshot(self, file_path: str = "storages/default/colony_snapshot.json"):
         self.logger.info(f"Loading network snapshot from {file_path}")
         nodes = []

colony/core/network/nodes.py

@@ -3,7 +3,6 @@ from typing import List
 
 from colony.core.network.hardware import HardwareStatistics
 from colony.core.network.requests import CommandRequest
-from colony.transport.whisperer import Whisperer
 
 
 class Node:
@@ -14,11 +13,6 @@ class Node:
         self.roles = roles
         self.port = port
 
-    def execute(self, command: Enum, *args, **kwargs):
-        request = CommandRequest.form(command, *args, **kwargs)
-
-        return Whisperer(self.ip, self.port).send(request.collapse())
-
     def to_dict(self):
         return {
             "id": self.id,
@@ -42,7 +36,7 @@ class LocalNode(Node):
     def set_role_instances(self, roles):
         self.__roles = dict(roles)
 
-    def execute_local(self, command: Enum, *args, **kwargs):
+    def execute_local(self, command: Enum, *args, caller_id=None, **kwargs):
         request = CommandRequest.form(command, *args, **kwargs)
 
         role = self.__roles.get(request.role)
@@ -52,5 +46,6 @@ class LocalNode(Node):
         return role.execute_local(
             request.id,
             *request.args,
+            caller_id=caller_id,
             **request.kwargs
         )

colony/core/network/requests.py

@@ -31,12 +31,13 @@ class CommandRequest:
     def form(cls, id: Enum, *args: Any, **kwargs: Any):
         return cls(id, args, kwargs)
 
-    def collapse(self):
-        return {
-            "command": {
-                "role": self.role,
-                "name": self.name,
-                "args": self.args,
-                "kwargs": self.kwargs
-            }
+    def collapse(self, sender=None):
+        command = {
+            "role": self.role,
+            "name": self.name,
+            "args": self.args,
+            "kwargs": self.kwargs
         }
+        if sender is not None:
+            command["from"] = sender
+        return {"command": command}

colony/core/roles/chatter/role.py

@@ -0,0 +1,41 @@
+from enum import Enum
+
+from colony.core.roles.role import Role, Command
+from colony.core.storages.chat import ChatStorage
+
+MAX_MESSAGE_LENGTH = 4000
+
+
+class Chatter(Role):
+    class COMMANDS(Enum):
+        RECEIVE = 'Chatter.Receive'
+
+    def __init__(self, context, storage=None):
+        commands = {
+            self.COMMANDS.RECEIVE.value: Command(self.receive, with_caller=True),
+        }
+
+        self.storage = storage or ChatStorage()
+
+        super().__init__(commands, context)
+
+    def send(self, peer_id, text):
+        node = self.context.node_by_id(peer_id)
+        if node is None or "Chatter" not in node.roles:
+            raise ValueError(f"{peer_id} is not running Chatter")
+
+        response = self.context.execute_on(node, self.COMMANDS.RECEIVE, text)
+        if response.get("status") != "ok":
+            raise ValueError(f"{peer_id} refused the message")
+
+        self.storage.append(peer_id, "out", text)
+
+    def receive(self, text, caller_id):
+        if not isinstance(text, str) or not text.strip() or len(text) > MAX_MESSAGE_LENGTH:
+            raise ValueError("Invalid message")
+
+        self.storage.append(caller_id, "in", text)
+        return True
+
+    def history(self, peer_id, after=0):
+        return self.storage.messages(peer_id)[after:]

colony/core/roles/manager.py

@@ -25,6 +25,19 @@ class RoleManager:
         if wanted != current:
             self.state_manager.set(StateManager.Data.Roles, wanted)
 
+    def ensure_on_demand_roles(self, role_names):
+        roles = self.role_registry.roles
+        on_demand = []
+        for role_name in role_names:
+            if role_name in roles and roles[role_name].on_demand:
+                on_demand.append(role_name)
+            else:
+                self.logger.warning(f"Ignoring {role_name} in roles.on_demand, it is not an on-demand role")
+        current = self.state_manager.get(StateManager.Data.Roles)
+        wanted = list(dict.fromkeys(current + on_demand))
+        if wanted != current:
+            self.state_manager.set(StateManager.Data.Roles, wanted)
+
     def initialize_roles(self):
         self.logger.info("Initializing roles...")
         context = RoleContext(self.network, self)

colony/core/roles/replicator/role.py

@@ -27,7 +27,6 @@ class Replicator(Role):
         super().__init__(commands, context)
 
     def start(self):
-        # catch up
         replicator_checksums = dict()
 
         for node in self.context.nodes_with_role(Replicator):

colony/core/roles/role.py

@@ -5,8 +5,9 @@ from colony.core.logger.logging import get_logger
 
 
 class Command:
-    def __init__(self, callback):
+    def __init__(self, callback, with_caller=False):
         self.callback = callback
+        self.with_caller = with_caller
 
 class Role:
     CONTINUOUS = None
@@ -56,10 +57,13 @@ class Role:
     def stop(self):
         self._stop_event.set()
 
-    def execute_local(self, name, *args, **kwargs):
+    def execute_local(self, name, *args, caller_id=None, **kwargs):
         command = self.commands.get(name)
 
         if not command:
             raise ValueError(f"No command {name}")
 
+        if command.with_caller:
+            kwargs["caller_id"] = caller_id
+
         return command.callback(*args, **kwargs)

colony/core/roles/webview/app.py

@@ -8,9 +8,13 @@ STATIC_DIR = Path(__file__).parent / "static"
 logging.getLogger("werkzeug").setLevel(logging.WARNING)
 
 
-def create_app(get_status):
+def create_app(get_status, get_chatter):
     app = Flask(__name__, static_folder=str(STATIC_DIR), static_url_path="/static")
 
+    def no_store(response):
+        response.headers["Cache-Control"] = "no-store"
+        return response
+
     @app.get("/")
     def index():
         return app.send_static_file("index.html")
@@ -23,8 +27,37 @@ def create_app(get_status):
             app.logger.exception("Could not build the status")
             return jsonify(error="status unavailable"), 500
 
-        response = jsonify(payload)
-        response.headers["Cache-Control"] = "no-store"
-        return response
+        return no_store(jsonify(payload))
+
+    @app.get("/api/chat")
+    def chat_history():
+        chatter = get_chatter()
+        if chatter is None:
+            return jsonify(error="Chatter is not running on this node"), 409
+
+        peer = request.args.get("peer")
+        after = request.args.get("after", "0")
+        if not peer or not after.isdigit():
+            return jsonify(error="peer and a numeric after are required"), 400
+
+        return no_store(jsonify(messages=chatter.history(peer, int(after))))
+
+    @app.post("/api/chat")
+    def chat_send():
+        chatter = get_chatter()
+        if chatter is None:
+            return jsonify(error="Chatter is not running on this node"), 409
+
+        body = request.get_json(silent=True)
+        if not isinstance(body, dict) or not isinstance(body.get("peer"), str) or not isinstance(body.get("text"), str):
+            return jsonify(error="peer and text are required"), 400
+
+        try:
+            chatter.send(body["peer"], body["text"])
+        except ValueError as e:
+            return jsonify(error=str(e)), 400
+        except OSError:
+            return jsonify(error=f"{body['peer']} cannot be reached"), 502
+        return jsonify(sent=True)
 
     return app

colony/core/roles/webview/role.py

@@ -21,7 +21,7 @@ class WebView(Role):
 
     def __init__(self, context):
         host, port = setting("web.host"), setting("web.port")
-        self.app = create_app(self._status)
+        self.app = create_app(self._status, self._chatter)
         try:
             self._server = make_server(host, port, self.app, threaded=True)
         except SystemExit:
@@ -39,6 +39,9 @@ class WebView(Role):
             raise RuntimeError("The Unit role is not available")
         return unit
 
+    def _chatter(self):
+        return self.context.self().get_role_instances().get("Chatter")
+
     def _status(self, compact=False):
         payload = self._unit().show_cluster()
         own_key = next((member["ip"], member["port"]) for member in payload["members"] if member["self"])

colony/core/roles/webview/static/app.js

@@ -528,6 +528,7 @@
     } else {
       drawStats();
     }
+    updateChat();
   }
 
   /* ----------------------------------------------------------------- controls */
@@ -597,6 +598,7 @@
     else announce("Selection cleared");
     drawRegions();
     drawTrace();
+    updateChat();
   }
 
   const same = (a, b) => Boolean(a && b && a.kind === b.kind && a.id === b.id);
@@ -848,6 +850,119 @@
     drawTrace();
   });
 
+  const [chatPanel, chatTitle, chatStatus, chatLog, chatEmpty, chatError, chatForm, chatInput, chatSend, chatClose, chatLive] = ["chat", "chat-title", "chat-status", "chat-log", "chat-empty", "chat-error", "chat-form", "chat-input", "chat-send", "chat-close", "chat-live"].map((id) => document.getElementById(id));
+  let chat = { peer: null, count: 0, loaded: false, loading: false };
+
+  const chatTarget = () => (model && selectionLocks() && selection.id !== model.selfKey ? model.byKey.get(selection.id) : null);
+  const chatTime = (at) => {
+    const date = new Date(at);
+    return isNaN(date) ? "" : date.toLocaleString([], { dateStyle: "short", timeStyle: "short" });
+  };
+
+  function updateChat() {
+    const peer = chatTarget();
+    chatPanel.hidden = !peer;
+    stage.classList.toggle("chatting", Boolean(peer));
+    if (!peer) {
+      chat = { peer: null, count: 0, loaded: false, loading: false };
+      return;
+    }
+    if (chat.peer !== peer.id) {
+      chat = { peer: peer.id, hostname: peer.hostname, count: 0, loaded: false, loading: false };
+      chatTitle.textContent = "Chat with " + peer.hostname;
+      chatLog.replaceChildren();
+      chatEmpty.hidden = true;
+      chatError.hidden = true;
+      chatInput.value = "";
+    }
+    const state = C.chatState(model, peer.key);
+    chatStatus.textContent = state.text;
+    chatStatus.hidden = !state.text;
+    chatForm.hidden = !state.send;
+    if (state.read) loadChat();
+  }
+
+  function paintChatMessages(messages, fresh) {
+    const atBottom = chatLog.scrollTop + chatLog.clientHeight >= chatLog.scrollHeight - 8;
+    messages.forEach((message) => {
+      const item = html("li", undefined, "msg " + (message.direction === "out" ? "out" : "in"));
+      item.appendChild(html("span", message.direction === "out" ? "You" : chat.hostname, "msg-from sr-only"));
+      item.appendChild(html("p", message.text, "msg-text"));
+      const time = html("time", chatTime(message.at), "msg-time");
+      time.dateTime = message.at;
+      item.appendChild(time);
+      chatLog.appendChild(item);
+    });
+    if (atBottom || !fresh) chatLog.scrollTop = chatLog.scrollHeight;
+    const incoming = messages.filter((message) => message.direction === "in");
+    if (fresh && incoming.length) chatLive.textContent = chat.hostname + " says: " + incoming[incoming.length - 1].text;
+  }
+
+  async function loadChat() {
+    const current = chat;
+    if (current.loading || !current.peer) return;
+    current.loading = true;
+    try {
+      const response = await fetch("/api/chat?peer=" + encodeURIComponent(current.peer) + "&after=" + current.count, { cache: "no-store" });
+      const body = await response.json().catch(() => ({}));
+      if (!response.ok) throw new Error(body.error || "the node answered with HTTP " + response.status);
+      if (chat !== current) return;
+      paintChatMessages(body.messages, current.loaded);
+      current.count += body.messages.length;
+      current.loaded = true;
+      chatEmpty.hidden = current.count > 0;
+      if (chatError.dataset.from === "load") chatError.hidden = true;
+    } catch (error) {
+      if (chat === current) {
+        chatError.textContent = "Cannot load the messages: " + error.message;
+        chatError.dataset.from = "load";
+        chatError.hidden = false;
+      }
+    } finally {
+      current.loading = false;
+    }
+  }
+
+  chatForm.addEventListener("submit", async (event) => {
+    event.preventDefault();
+    const text = chatInput.value.trim();
+    const current = chat;
+    if (!text || !current.peer || chatSend.disabled) return;
+    chatInput.disabled = chatSend.disabled = true;
+    try {
+      const response = await fetch("/api/chat", {
+        method: "POST",
+        headers: { "Content-Type": "application/json" },
+        body: JSON.stringify({ peer: current.peer, text: text }),
+      });
+      const body = await response.json().catch(() => ({}));
+      if (!response.ok) throw new Error(body.error || "the node answered with HTTP " + response.status);
+      if (chat === current) {
+        chatInput.value = "";
+        chatError.hidden = true;
+        loadChat();
+      }
+    } catch (error) {
+      if (chat === current) {
+        chatError.textContent = "Not sent: " + error.message;
+        chatError.dataset.from = "send";
+        chatError.hidden = false;
+      }
+    } finally {
+      chatInput.disabled = chatSend.disabled = false;
+      if (!chatPanel.hidden) chatInput.focus();
+    }
+  });
+
+  chatClose.addEventListener("click", () => select(null));
+  chatPanel.addEventListener("keydown", (event) => {
+    if (event.key !== "Escape") return;
+    event.preventDefault();
+    const key = selection && selection.id;
+    select(null);
+    if (key && model.byKey.has(key)) focusNode(key);
+  });
+
   /* --------------------------------------------------------------------- data */
 
   if (typeof ResizeObserver !== "undefined") new ResizeObserver(resize).observe(stage);

colony/core/roles/webview/static/index.html

@@ -65,7 +65,24 @@
     </div>
     <button id="legend-open" type="button" class="legend-open" aria-label="Show the legend" title="Show the legend" hidden>?</button>
     <p id="updated" class="muted stamp"></p>
+
+    <aside id="chat" class="chat" aria-labelledby="chat-title" hidden>
+      <div class="chat-head">
+        <h2 id="chat-title"></h2>
+        <button id="chat-close" type="button" aria-label="Close the chat" title="Close the chat (Escape)"><svg viewBox="0 0 20 20" aria-hidden="true"><path d="M5 5l10 10M15 5L5 15"/></svg></button>
+      </div>
+      <p id="chat-status" class="chat-status" hidden></p>
+      <ol id="chat-log" class="chat-log" role="log" aria-label="Messages" tabindex="0"></ol>
+      <p id="chat-empty" class="chat-empty muted" hidden>No messages yet.</p>
+      <p id="chat-error" class="error chat-error" role="alert" hidden></p>
+      <form id="chat-form" class="chat-form">
+        <label for="chat-input" class="sr-only">Message</label>
+        <input id="chat-input" type="text" maxlength="4000" autocomplete="off" placeholder="Message">
+        <button id="chat-send" type="submit">Send</button>
+      </form>
+    </aside>
   </main>
+  <div id="chat-live" class="sr-only" aria-live="polite"></div>
 
   <div id="live" class="sr-only" role="status" aria-live="polite"></div>
   <ul id="reader" class="sr-only" aria-label="Networks and their nodes"></ul>

colony/core/roles/webview/static/model.js

@@ -101,6 +101,7 @@
       index: index,
       key: address(member),
       net: networkOf(member.ip),
+      id: member.hostname + "@" + member.ip,
       hostname: member.hostname,
       ip: member.ip,
       port: member.port,
@@ -236,6 +237,18 @@
     return off.length ? "Not running (on demand): " + off.join(", ") + " · start one with: colonyctl role add " + off[0] : "";
   }
 
+  function chatState(model, key) {
+    const peer = model.byKey.get(key);
+    if (!peer || peer.self) return null;
+    const own = model.byKey.get(model.selfKey);
+    if (!own || !own.roles.includes("Chatter")) {
+      return { read: false, send: false, text: "Chat needs Chatter on this node too. Start it with: colonyctl role add Chatter" };
+    }
+    if (!peer.roles.includes("Chatter")) return { read: true, send: false, text: peer.hostname + " is not running Chatter, so it cannot receive messages." };
+    if (!peer.alive) return { read: true, send: false, text: peer.hostname + " is down." };
+    return { read: true, send: true, text: "" };
+  }
+
   function summarize(model) {
     const alive = model.nodes.filter((node) => node.alive).length;
     const down = model.nodes.length - alive;
@@ -367,7 +380,7 @@
     return lines.join("\n");
   }
 
-  const api = { describeStats, describeRows, networkStats, plural, barLevel, OTHER_NETWORK, COLORS, address, keyOf, networkOf, compareNetworks, compareAddresses, shorten, wrapLabels, analyze, bridgesOf, resolveView, routesFrom, routeSummary, specialRoles, alerts, onDemandHint, summarize, describeNode };
+  const api = { describeStats, describeRows, networkStats, plural, barLevel, OTHER_NETWORK, COLORS, address, keyOf, networkOf, compareNetworks, compareAddresses, shorten, wrapLabels, analyze, bridgesOf, resolveView, routesFrom, routeSummary, specialRoles, alerts, onDemandHint, chatState, summarize, describeNode };
   if (typeof module !== "undefined") module.exports = api;
   else globalThis.Colony = Object.assign(globalThis.Colony || {}, api);
 })();

colony/core/roles/webview/static/style.css

@@ -216,6 +216,32 @@ p { margin: 4px 0; }
 .swatch.flow { width: 20px; height: 0; border-radius: 0; border: 0; border-top: 3.5px dashed var(--text); vertical-align: 3px; }
 .swatch.none { background: transparent; border: 2px dashed var(--muted); }
 
+.chat { position: absolute; top: 12px; right: 12px; bottom: 12px; width: min(340px, calc(100% - 24px)); display: flex; flex-direction: column; border: 1px solid var(--line); border-radius: 10px; background: var(--card); box-shadow: 0 6px 24px color-mix(in srgb, var(--text) 12%, transparent); }
+.chat-head { display: flex; align-items: center; gap: 8px; padding: 8px 8px 8px 14px; border-bottom: 1px solid var(--line); }
+.chat-head h2 { flex: 1; margin: 0; font-size: 15px; overflow: hidden; text-overflow: ellipsis; white-space: nowrap; }
+.chat-head button { display: inline-flex; align-items: center; justify-content: center; min-width: 32px; height: 32px; padding: 0; border: 0; border-radius: 8px; background: transparent; color: var(--text); cursor: pointer; }
+.chat-head button:hover { background: color-mix(in srgb, var(--text) 8%, transparent); }
+.chat-head svg { width: 18px; height: 18px; fill: none; stroke: currentColor; stroke-width: 1.8; stroke-linecap: round; }
+.chat-status { margin: 0; padding: 8px 14px; border-bottom: 1px solid var(--line); color: var(--muted); font-size: 13px; }
+.chat-log { flex: 1; min-height: 0; overflow-y: auto; margin: 0; padding: 12px; list-style: none; display: flex; flex-direction: column; gap: 6px; }
+.chat-log:focus-visible { outline: 2px solid var(--text); outline-offset: -2px; }
+.chat-empty { margin: 0; padding: 0 14px 12px; text-align: center; }
+.chat-error { margin: 0; padding: 6px 14px; font-size: 13px; }
+.msg { max-width: 85%; padding: 6px 10px; border-radius: 12px; overflow-wrap: anywhere; }
+.msg.in { align-self: flex-start; background: var(--bg); border: 1px solid var(--line); border-bottom-left-radius: 4px; }
+.msg.out { align-self: flex-end; background: color-mix(in srgb, var(--n0) 16%, var(--card)); border: 1px solid color-mix(in srgb, var(--n0) 40%, transparent); border-bottom-right-radius: 4px; }
+.msg-text { margin: 0; white-space: pre-wrap; }
+.msg-time { display: block; margin-top: 2px; color: var(--muted); font-size: 11px; font-variant-numeric: tabular-nums; }
+.msg.out .msg-time { text-align: right; }
+.chat-form { display: flex; gap: 6px; padding: 10px; border-top: 1px solid var(--line); }
+.chat-form input { flex: 1; min-width: 0; height: 36px; padding: 0 10px; border: 1px solid var(--line); border-radius: 8px; background: var(--card); color: var(--text); font: inherit; }
+.chat-form input:focus-visible { outline: 2px solid var(--n0); outline-offset: 0; border-color: transparent; }
+.chat-form button { height: 36px; padding: 0 14px; border: 0; border-radius: 8px; background: var(--n0); color: var(--card); font: inherit; font-weight: 600; cursor: pointer; }
+.chat-form button:disabled, .chat-form input:disabled { opacity: 0.6; cursor: default; }
+.chat-form button:focus-visible { outline: 2px solid var(--text); outline-offset: 2px; }
+#stage.chatting .controls { right: 364px; }
+@media (pointer: coarse) { .chat-form input, .chat-form button, .chat-head button { min-height: 44px; } .chat-head button { min-width: 44px; } }
+
 .sr-only { position: absolute; width: 1px; height: 1px; margin: -1px; padding: 0; overflow: hidden; clip: rect(0 0 0 0); white-space: nowrap; border: 0; }
 
 /* Phones: the desktop sizes are small on a 400px screen, so the page text and touch targets grow. */
@@ -226,4 +252,6 @@ p { margin: 4px 0; }
   .alert, .route, .muted, .legend { font-size: 14px; }
   .legend { max-width: calc(100% - 24px); }
   .legend, .legend-open { bottom: 64px; }
+  .chat { top: auto; left: 0; right: 0; bottom: 0; width: auto; height: min(70%, 520px); border-radius: 14px 14px 0 0; border-bottom: 0; }
+  #stage.chatting .controls, #stage.chatting .legend, #stage.chatting .legend-open { display: none; }
 }

colony/core/storages/chat.py

@@ -0,0 +1,29 @@
+import datetime
+import os
+import threading
+from urllib.parse import quote
+
+from colony.core.storages.core import Storage, STORAGES_DIR
+
+CHAT_DIR = os.path.join(STORAGES_DIR, "chat")
+
+
+class ChatStorage(Storage):
+    _lock = threading.Lock()
+
+    def __init__(self, storage_dir=None):
+        super().__init__(storage_dir or CHAT_DIR)
+
+    @staticmethod
+    def _file(peer_id):
+        return quote(peer_id, safe="") + ".json"
+
+    def messages(self, peer_id):
+        return self.open(self._file(peer_id)) or []
+
+    def append(self, peer_id, direction, text):
+        at = datetime.datetime.now(datetime.timezone.utc).isoformat()
+        with self._lock:
+            messages = self.messages(peer_id)
+            messages.append({"direction": direction, "text": text, "at": at})
+            self.save(self._file(peer_id), messages)

colony/settings.py

@@ -41,9 +41,11 @@ DEFAULTS = {
         "Replicator.HashSum",
         "Replicator.Retrieve",
         "Replicator.Replicate",
+        "Chatter.Receive",
     ],
     "web.host": "0.0.0.0",
     "web.port": 8080,
+    "roles.on_demand": [],
     "role_policy.max_roles": None,
     "role_policy.deny": [],
     "role_policy.min_scores": {},

config.json

@@ -58,9 +58,13 @@
       "Unit.RequestRole",
       "Replicator.HashSum",
       "Replicator.Retrieve",
-      "Replicator.Replicate"
+      "Replicator.Replicate",
+      "Chatter.Receive"
     ]
   },
+  "roles": {
+    "on_demand": []
+  },
   "role_policy": {
     "max_roles": null,
     "deny": [],

docker-compose.yml

@@ -52,6 +52,8 @@ services:
         "Unit.ShowCluster","Unit.RequestRole","Replicator.HashSum",
         "Replicator.Retrieve","Replicator.Replicate"]
 
+      COLONY_ROLES__ON_DEMAND: "[]"
+
       COLONY_ROLE_POLICY__MAX_ROLES: null
       COLONY_ROLE_POLICY__DENY: "[]"
       COLONY_ROLE_POLICY__MIN_SCORES: "{}"

main.py

@@ -39,6 +39,7 @@ def main():
     )
 
     role_manager.ensure_general_roles()
+    role_manager.ensure_on_demand_roles(setting("roles.on_demand"))
     role_manager.initialize_roles()
 
     listener = Listener(dispatcher=dispatcher)

role_settings.py

@@ -1,3 +1,4 @@
+from colony.core.roles.chatter.role import Chatter
 from colony.core.roles.explorer.role import Explorer
 from colony.core.roles.replicator.role import Replicator
 from colony.core.roles.unit.role import Unit
@@ -59,5 +60,10 @@ available_roles = [
             }
         ),
         minimal_percentage=51
+    ),
+    RoleStats(
+        cls=Chatter,
+        priority=RolePriority.LOW,
+        on_demand=True,
     )
 ]