cleanup

.claude/project-map.md

@@ -0,0 +1,58 @@
+# Project map: colony
+
+_Auto-generated by the `explorer` subagent. Factual overview — refresh via
+explorer, not by hand._
+
+## What it does
+Self-organizing cluster of nodes on a LAN (README.md). Nodes discover each other (TCP probe on the discovery port over a configured IP range), gossip membership, and bid for "roles" (replicated storage, password vault, web dashboard, chat, ...), so roles keep running as nodes join/leave/die. Each node is one Python process; `colonyctl` controls the local node.
+
+## How it works
+- `main.py` boot order: validate discovery range -> `StateManager` (data/state.json) -> `Network` -> `RoleRegistry(available_roles)` -> `Dispatcher` + `RoleManager` (general + on-demand roles, init) -> TLS `Listener` bind -> unix `ControlSocket` bind -> threads for ControlSocket, Listener, `Discover.listen` -> `Gossip.join_or_bootstrap()`; if bootstrapped, `Foreman.tick()` runs to assign roles -> `start_all()` -> `ControlQueue.run_forever()` on main thread.
+- Node-to-node: length-prefixed (4-byte `!I`, 16 MiB max) JSON frames over TLS (`colony/transport/framing.py`, `listener.py`, `whisperer.py` client). Payload `{"command": {role, name, args, kwargs, from}}`; `Dispatcher.dispatch` maps `Role.Command` to the role's `COMMANDS` enum and calls `execute_local`.
+- Roles (`colony/core/roles/<name>/role.py`) subclass `Role` (`roles/role.py`): `CONTINUOUS` + `LOOP_SECONDS_WAIT` run `loop()` in a daemon thread; commands registered via `Command(callback, with_caller)`.
+- Role assignment: `Foreman` proposes unclaimed roles to capable nodes; `RoleManager` (`roles/manager.py`) + `policy.py` (config `role_policy`) + `role_registry.py` (priority, hardware preferences, `minimal_percentage`, `on_demand`) decide who runs what. Hardware scoring: `core/hardware_scores.py`, `network/hardware.py`.
+- Unreachable nodes: `Gateway` role relays commands (`network/relay.py`, builds `Gateway.Forward`; config `gateway.allow_commands` whitelists relayable commands).
+- Local control: `colonyctl` -> unix socket `data/colony.sock` (`transport/control.py`) -> `ControlQueue` -> `ControlActions` (`core/control.py`).
+
+## Tech stack
+Python >=3.14 (`.python-version` 3.14), uv (`uv.lock`, `package = false`). Deps (pyproject.toml): dependency-injector 4.48.3, dynaconf 3.2.12, Flask 3.1.3, GPUtil 1.4.0, psutil 7.2.1, setuptools 80.9.0. Web UI static JS/CSS (no build step). Docker image python:3.14-alpine.
+
+## Repositories / modules
+Single flat package.
+- `main.py` node entrypoint; `colonyctl.py` -> `colony/cli.py` (argparse CLI); `role_settings.py` role list; `config.json` defaults.
+- `colony/core/` — `context.py` (RoleContext), `dispatcher.py`, `control.py`, `state_manager.py`, `logger/`, `exceptions.py`.
+- `colony/core/network/` — `core.py` (Network), `membership.py`, `nodes.py`, `advice.py`, `hardware.py`, `relay.py`, `requests.py`.
+- `colony/core/roles/` — Unit, Gossip, Foreman, Gateway, VaultWarden, Replicator, Explorer, WebView (Flask `app.py` + `static/`), Chatter; `manager.py`, `policy.py`.
+- `colony/core/storages/` — `core.py` (Storage base, `storages/` dir, tar/hash helpers), `default.py`, `replica.py`, `chat.py`.
+- `colony/transport/` — `listener.py`, `whisperer.py`, `discover.py`, `control.py`, `framing.py`.
+- `colony/utils/` — `files.py` (atomic_write_json), `network.py` (hosts_in_range).
+- `certs/`, `data/`, `storages/`, `logs/` — runtime dirs (git-ignored: certs, data, storages).
+
+## Key implementations
+- Startup/wiring: `main.py`.
+- Role catalogue and priorities: `role_settings.py` (Unit, Gossip, Foreman, Gateway = GENERAL; VaultWarden, Replicator = CRITICAL (Replicator `minimal_percentage=51`); Explorer = HIGH; WebView, Chatter = LOW and on-demand).
+- Role base/lifecycle: `colony/core/roles/role.py`, `manager.py`.
+- Membership/join: `roles/gossip/role.py`, `network/membership.py`.
+- Role placement: `roles/foreman/role.py`, `roles/policy.py`.
+- CLI (`status`, `role add`, `webview`, `chat`): `colony/cli.py`.
+- Settings: `colony/settings.py` (`setting("a.b")` raises KeyError when missing).
+
+## Build, run, test
+- Install/run: `uv sync` then `uv run python main.py`.
+- CLI: `./colonyctl.py status [--json]`, `role add <Role>`, `webview`, `chat <node> [text]`.
+- Docker: `docker build -t colony .`; `docker compose up -d` (host network; volumes /app/data, /app/storages, /app/logs; ports 65334/tcp, 28149/tcp, 8080/tcp).
+- CI: `.gitea/workflows/docker.yml` builds and pushes `registry.ciomek.pl/colony:latest` on push to main.
+- Tests: none in the repo.
+
+## External integrations
+No third-party services. Peers over TLS (cert/key at `certs/server.crt`, `certs/server.key`), TCP discovery on 28149, optional web dashboard on 8080 with basic credentials (`web.username/password`), private Docker registry in CI.
+
+## Conventions & gotchas
+- Config: `config.json` overridden by env `COLONY_<SECTION>__<KEY>` (dynaconf); `docker-compose.yml` lists keys.
+- Discovery range defaults to 192.168.65.100-120; node exits if range invalid, listener or control socket cannot bind. Warns if node IP is 127.x.
+- Role commands (`COMMANDS` enums) are for node-to-node calls; the CLI uses the control socket actions instead.
+- State persisted atomically in `data/state.json`; a corrupt file is copied to `.corrupt` and reset.
+- Loop exceptions in continuous roles are logged and retried.
+- Roles with `on_demand=True` only start via `roles.on_demand` config or `colonyctl role add`.
+
+<!-- generated-at-commit: 244fefd8530bf5cd01267a13f8aee34782c31934 | 2026-10-06 -->

Dockerfile

@@ -23,6 +23,6 @@ RUN chmod +x colonyctl.py && ln -s /app/colonyctl.py /usr/local/bin/colonyctl
 
 VOLUME ["/app/data", "/app/storages", "/app/logs"]
 
-EXPOSE 65334/tcp 28149/udp 8080/tcp
+EXPOSE 65334/tcp 28149/tcp 8080/tcp
 
 CMD ["python", "main.py"]

README.md

@@ -15,7 +15,7 @@ cluster keeps running the roles it needs even as nodes join, leave, or die.
   hardware preferences (disk, longevity, etc.) and policy (`role_policy` in
   `config.json`).
 - **Transport**: nodes talk over a TLS listener (`Listener`/`Whisperer`) and
-  discover each other over UDP (`Discover`). A local control socket
+  discover each other over TCP (`Discover`). A local control socket
   (`ControlSocket`) is how `colonyctl` talks to the node process on the same
   host.
 - **Roles shipped today** (see `role_settings.py`):
@@ -46,6 +46,14 @@ Configuration lives in `config.json` and can be overridden with
 every available key). At minimum, point `discovery.range_start`/`range_end`
 at the IP range your nodes live on.
 
+Certificates are loaded once per process:
+restart the node after rotating them. To set a key to null through the
+environment, use `@none` (a plain `null` is read as a string).
+
+Replicated storage (the vault and its replicas) travels as one archive, so it
+is limited to 16 MiB per node, and to `gateway.max_payload_bytes` (4 MiB by
+default) when a replica is fetched through a gateway.
+
 ## Controlling a node (`colonyctl`)
 
 `colonyctl` talks to the local node over its control socket:

colony/cli.py

@@ -4,7 +4,6 @@ import sys
 import time
 from datetime import datetime
 
-from colony.core.storages.chat import ChatStorage
 from colony.settings import setting
 from colony.transport import control
 
@@ -26,13 +25,28 @@ def call(args, action, **params):
 
 
 def table(rows):
-    widths = [max(len(str(row[i])) for row in rows) for i in range(len(rows[0]))]
-    return "\n".join("  ".join(str(cell).ljust(width) for cell, width in zip(row, widths)).rstrip() for row in rows)
+    widths = []
+    for column in range(len(rows[0])):
+        width = 0
+        for row in rows:
+            width = max(width, len(str(row[column])))
+        widths.append(width)
+
+    lines = []
+    for row in rows:
+        cells = []
+        for cell, width in zip(row, widths):
+            cells.append(str(cell).ljust(width))
+        lines.append("  ".join(cells).rstrip())
+    return "\n".join(lines)
 
 
 def usage(stats, name):
-    used, total = stats.get(f"{name}_used_gb"), stats.get(f"{name}_total_gb")
-    return f"{used:.1f}/{total:.1f} GB" if used is not None and total is not None else "-"
+    used = stats.get(f"{name}_used_gb")
+    total = stats.get(f"{name}_total_gb")
+    if used is None or total is None:
+        return "-"
+    return f"{used:.1f}/{total:.1f} GB"
 
 
 def render_status(cluster):
@@ -62,16 +76,33 @@ def render_status(cluster):
 def render_chat(messages, peer):
     if not messages:
         return f"No messages with {peer} yet."
-    return "\n".join(f"{datetime.fromisoformat(m['at']).astimezone():%H:%M}  {'you' if m['direction'] == 'out' else peer}: {m['text']}" for m in messages)
+
+    lines = []
+    for message in messages:
+        sent_at = datetime.fromisoformat(message["at"]).astimezone()
+        author = peer
+        if message["direction"] == "out":
+            author = "you"
+        lines.append(f"{sent_at:%H:%M}  {author}: {message['text']}")
+    return "\n".join(lines)
 
 
 def find_node(cluster, name):
-    matches = [
-        member for member in cluster["members"]
-        if name.lower() in (member["hostname"].lower(), f"{member['ip']}:{member['port']}", f"{member['hostname']}@{member['ip']}".lower())
-    ]
-    if len(matches) != 1:
-        raise CliError(f"{'no node' if not matches else 'more than one node'} matches {name!r}")
+    wanted = name.lower()
+    matches = []
+    for member in cluster["members"]:
+        names = [
+            member["hostname"].lower(),
+            f"{member['ip']}:{member['port']}",
+            f"{member['hostname']}@{member['ip']}".lower(),
+        ]
+        if wanted in names:
+            matches.append(member)
+
+    if len(matches) == 0:
+        raise CliError(f"no node matches {name!r}")
+    if len(matches) > 1:
+        raise CliError(f"more than one node matches {name!r}")
     return matches[0]
 
 
@@ -122,8 +153,12 @@ def chat(args):
         call(args, "send_chat", peer=peer_id, text=args.text)
         print(f"Sent to {peer['hostname']}")
     else:
-        history = lambda: ChatStorage().messages(peer_id)
-        show(args, lambda: json.dumps(history(), indent=2) if args.json else render_chat(history(), peer["hostname"]))
+        def render():
+            history = call(args, "chat_history", peer=peer_id)
+            if args.json:
+                return json.dumps(history, indent=2)
+            return render_chat(history, peer["hostname"])
+        show(args, render)
 
 
 def build_parser():

colony/core/control.py

@@ -5,6 +5,7 @@ class ControlActions:
             "status": self.status,
             "add_role": self.add_role,
             "send_chat": self.send_chat,
+            "chat_history": self.chat_history,
         }
 
     def run(self, payload):
@@ -28,7 +29,12 @@ class ControlActions:
         return self._role("Unit").show_cluster()
 
     def add_role(self, role):
-        return self._role("Unit").request_role(role)
+        return self._role("Unit").add_role_locally(role)
 
     def send_chat(self, peer, text):
         self._role("Chatter").send(peer, text)
+
+    def chat_history(self, peer):
+        if not isinstance(peer, str):
+            raise ValueError("peer must be a node id")
+        return self._role("Chatter").history(peer)

colony/core/exceptions.py

@@ -1,5 +0,0 @@
-class NetworkException(Exception):
-    pass
-
-class NoRoleException(Exception):
-    pass

colony/core/network/core.py

@@ -1,4 +1,3 @@
-import json
 from enum import Enum
 from typing import List, Optional
 
@@ -186,52 +185,3 @@ class Network:
         self._via.pop(key, None)
         self._direct_ok[key] = self.membership.now()
         return response
-
-    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 = []
-        try:
-            with open(file_path, "r", encoding="utf-8") as f:
-                snapshot = json.load(f)
-        except FileNotFoundError:
-            self.logger.warning(f"Snapshot file {file_path} not found. Starting with empty network")
-            return nodes
-        except (json.JSONDecodeError, OSError) as e:
-            self.logger.warning(f"Could not read snapshot {file_path}: {e}")
-            return nodes
-
-        if not isinstance(snapshot, dict):
-            self.logger.warning(f"Snapshot {file_path} is not an object. Ignoring it")
-            return nodes
-
-        for node_ip, node_data in snapshot.items():
-            if not isinstance(node_data, dict):
-                self.logger.warning(f"Skipping node {node_ip}: entry is not a dict")
-                continue
-
-            if node_data.get("status") == "unreachable":
-                self.logger.info(f"Skipping node {node_ip}: last seen unreachable")
-                continue
-
-            result = node_data.get("result")
-
-            if not isinstance(result, dict):
-                self.logger.warning(f"Skipping node {node_ip}: result is not a dict")
-                continue
-
-            roles = result.get("roles", [])
-            try:
-                node = Node(
-                    hostname=result["hostname"],
-                    ip=result["ip"],
-                    port=result["port"],
-                    roles=roles if isinstance(roles, list) else []
-                )
-            except KeyError as e:
-                self.logger.warning(f"Skipping node {node_ip}: result has no {e}")
-                continue
-
-            self.logger.info(f"Loaded node from snapshot: {node.id}")
-            nodes.append(node)
-
-        return nodes

colony/core/network/hardware.py

@@ -1,6 +1,5 @@
 import datetime
 import platform
-import json
 
 import GPUtil
 import psutil
@@ -74,23 +73,6 @@ class HardwareStatistics:
             stats["uptime_hours"] = HardwareScorer.longevity_score(self)
         return stats
 
-    def to_json(self):
-        return json.dumps(self.to_dict(), indent=2)
-
-    @staticmethod
-    def from_dict(hw_dict: dict):
-        hw = HardwareStatistics()
-        hw.cpu = hw_dict.get("cpu")
-        hw.mem = hw_dict.get("memory")
-        hw.gpu = hw_dict.get("gpu")
-        hw.disk = hw_dict.get("disk")
-        hw.os_info = hw_dict.get("os")
-        hw.boot_time = hw_dict.get("boot_time")
-        # Optionally parse boot_time back to datetime
-        if hw.boot_time:
-            hw.boot_time = datetime.datetime.fromisoformat(hw.boot_time)
-        return hw
-
     def _get_cpu_stats(self):
         return {
             "cores": psutil.cpu_count(logical=False),

colony/core/network/membership.py

@@ -135,6 +135,10 @@ class Membership:
     def key(self):
         return self._self.key
 
+    @property
+    def max_records(self):
+        return self.max_members + 1
+
     def now(self):
         return self.clock()
 
@@ -197,13 +201,29 @@ class Membership:
                 current = self._entries.get(record.key)
                 if current is not None and record.version <= current[0].version:
                     continue
-                if current is None and len(self._entries) >= self.max_members:
-                    continue
+                if current is None:
+                    if not self.admit(record):
+                        continue
+                    if len(self._entries) >= self.max_members and not self._evict_dead(now):
+                        continue
                 self._entries[record.key] = (record, now)
                 changed += 1
             self._purge(now)
         return changed
 
+    def _evict_dead(self, now):
+        oldest_key = None
+        oldest_age = self.member_timeout
+        for key, (_, advanced) in self._entries.items():
+            age = now - advanced
+            if age > oldest_age:
+                oldest_key = key
+                oldest_age = age
+        if oldest_key is None:
+            return False
+        del self._entries[oldest_key]
+        return True
+
     def _purge(self, now):
         for key in [key for key, (_, advanced) in self._entries.items() if now - advanced > self.forget_after]:
             del self._entries[key]

colony/core/roles/chatter/role.py

@@ -35,6 +35,10 @@ class Chatter(Role):
     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")
+        if not caller_id:
+            raise ValueError("Missing sender")
+        if self.context.node_by_id(caller_id) is None:
+            raise ValueError("Unknown sender")
 
         self.storage.append(caller_id, "in", text, self.max_messages)
         return True

colony/core/roles/explorer/role.py

@@ -155,7 +155,7 @@ class Explorer(Role):
             return []
 
         members = []
-        for row in rows[:setting("membership.max_members")]:
+        for row in rows[:self.context.membership().max_records]:
             try:
                 record = MemberRecord.from_dict(row)
             except ValueError as e:
@@ -216,5 +216,5 @@ class Explorer(Role):
         except (ValueError, TypeError) as e:
             self.logger.warning(f"Ignoring bad member list from {ip}:{port}: {e}")
             return False
-        membership.merge(members[:setting("membership.max_members")])
+        membership.merge(members[:membership.max_records])
         return True

colony/core/roles/gossip/role.py

@@ -43,7 +43,6 @@ class Gossip(Role):
         self.reprobe_interval = setting("membership.reprobe_interval")
         self.fanout = setting("membership.fanout")
         self.hello_timeout = setting("membership.hello_timeout")
-        self.max_members = setting("membership.max_members")
         self.discover_port = setting("discovery.discover_port")
         self.probe_timeout = setting("discovery.probe_timeout")
         self.probe_workers = setting("discovery.probe_workers")
@@ -73,10 +72,13 @@ class Gossip(Role):
 
         records = self._parse_records(known)
         self.membership.merge([record] + records)
+        if not self.membership.knows(record.key):
+            self.logger.debug(f"No room for {record.ip}:{record.port}, not admitting it")
+            return {"admitted": False, "members": []}
         return {"admitted": True, "members": self.membership.export(), "responder": self.membership.self_record().to_dict()}
 
     def _parse_records(self, raw):
-        if not isinstance(raw, list) or len(raw) > self.max_members:
+        if not isinstance(raw, list) or len(raw) > self.membership.max_records:
             raise ValueError("Member list must be a list of bounded size")
         return [MemberRecord.from_dict(item) for item in raw]
 

colony/core/roles/manager.py

@@ -96,12 +96,7 @@ class RoleManager:
                 self.logger.info("Roles already up to date")
                 return
 
-            context = RoleContext(self.network, self)
-            added = {}
-            for role_name in roles_to_add:
-                self.logger.debug(f"Creating new role {role_name}")
-                role_cls = self.role_registry.get_role_class_by_name(role_name)
-                added[role_name] = role_cls(context=context)
+            added = self._create_roles(roles_to_add)
 
             final_roles = {
                 name: current_roles[name] if name in current_roles else added[name]
@@ -111,15 +106,45 @@ class RoleManager:
             self.state_manager.set(StateManager.Data.Roles, list(final_roles.keys()))
             self._publish(final_roles)
 
-            for role_name in roles_to_remove:
-                self.logger.debug(f"Stopping removed role {role_name}")
-                current_roles[role_name].stop()
+            self._start_roles(added, current_roles)
 
+            removed = {name: current_roles[name] for name in roles_to_remove}
+            self._stop_roles(removed)
+
+        self.logger.info(f"Roles updated to: {list(final_roles.keys())}")
+
+    def _create_roles(self, role_names):
+        context = RoleContext(self.network, self)
+        added = {}
+        try:
+            for role_name in role_names:
+                self.logger.debug(f"Creating new role {role_name}")
+                role_cls = self.role_registry.get_role_class_by_name(role_name)
+                added[role_name] = role_cls(context=context)
+        except Exception:
+            self._stop_roles(added)
+            raise
+        return added
+
+    def _start_roles(self, added, previous_roles):
+        try:
             for role_name, role in added.items():
                 self.logger.info(f"Starting role {role_name}")
                 role.start()
+        except Exception:
+            self.logger.warning("A new role failed to start, going back to the previous roles")
+            self._stop_roles(added)
+            self.state_manager.set(StateManager.Data.Roles, list(previous_roles.keys()))
+            self._publish(previous_roles)
+            raise
 
-        self.logger.info(f"Roles updated to: {list(final_roles.keys())}")
+    def _stop_roles(self, roles):
+        for role_name, role in roles.items():
+            self.logger.debug(f"Stopping role {role_name}")
+            try:
+                role.stop()
+            except Exception:
+                self.logger.exception(f"Could not stop role {role_name}")
 
     def drop_roles(self, role_names):
         with self._lock:
@@ -132,7 +157,7 @@ class RoleManager:
     def _decline(self, role_name):
         self.network.membership.decline(role_name, self.role_policy.decline_cooldown)
 
-    def request_role(self, role_name):
+    def request_role(self, role_name, allow_on_demand=False):
         membership = self.network.membership
 
         with self._lock:
@@ -143,6 +168,9 @@ class RoleManager:
             if role_name in held:
                 return {"accepted": True, "reason": "already held"}
 
+            if self.role_registry.roles[role_name].on_demand and not allow_on_demand:
+                return {"accepted": False, "reason": "on-demand roles can only be added on the node itself"}
+
             if membership.is_declined(role_name):
                 return {"accepted": False, "reason": "recently declined"}
 

colony/core/roles/policy.py

@@ -15,11 +15,23 @@ class RolePolicy:
 class DefaultRolePolicy(RolePolicy):
     def __init__(self, max_roles=None, deny=None, min_scores=None, decline_cooldown=None):
         self.max_roles = setting("role_policy.max_roles", max_roles)
+        self._validate_max_roles()
         self.deny = set(setting("role_policy.deny", deny))
         raw_minimums = setting("role_policy.min_scores", min_scores)
         self.min_scores = {role: dict(minimums) for role, minimums in dict(raw_minimums).items()}
         self.decline_cooldown = setting("role_policy.decline_cooldown", decline_cooldown)
 
+    def _validate_max_roles(self):
+        if self.max_roles is None:
+            return
+        error = ValueError(f"role_policy.max_roles must be a whole number or null, got {self.max_roles!r}")
+        if isinstance(self.max_roles, bool):
+            raise error
+        if not isinstance(self.max_roles, int):
+            raise error
+        if self.max_roles < 0:
+            raise error
+
     def _violation(self, role_name, scores):
         if role_name in self.deny:
             return "role is on the deny list"

colony/core/roles/replicator/role.py

@@ -1,10 +1,75 @@
 import base64
 import random
+import threading
 from collections import Counter
 from enum import Enum
 
+from colony.core.logger.logging import get_logger
 from colony.core.roles.role import Role, Command
 from colony.core.storages.replica import ReplicaStorage
+from colony.transport.framing import MAX_FRAME_SIZE
+
+HASH_TIMEOUT = 3
+RETRIEVE_TIMEOUT = 30
+FRAME_HEADROOM = 4096
+
+logger = get_logger("Replicator")
+
+
+def reply_result(response):
+    if isinstance(response, dict) and response.get("status") == "ok":
+        return response.get("result")
+    return None
+
+
+def majority_copy(context, local_checksum=None):
+    own_key = context.membership().key
+    checksums = {}
+
+    for node in context.nodes_with_role(Replicator):
+        if (node.ip, node.port) == own_key:
+            continue
+        try:
+            response = context.execute_on(node, Replicator.COMMANDS.HASH_SUM, timeout=HASH_TIMEOUT)
+        except Exception as e:
+            logger.warning(f"Could not get the replica checksum from {node.ip}:{node.port}: {e}")
+            continue
+        checksum = reply_result(response)
+        if isinstance(checksum, str):
+            checksums[node] = checksum
+
+    votes = list(checksums.values())
+    if local_checksum is not None:
+        votes.append(local_checksum)
+    if not votes:
+        return None, []
+
+    checksum = Counter(votes).most_common(1)[0][0]
+    matching = [node for node, node_checksum in checksums.items() if node_checksum == checksum]
+    return checksum, matching
+
+
+def fetch_copy(context, nodes):
+    nodes = list(nodes)
+    random.shuffle(nodes)
+    for node in nodes:
+        try:
+            response = context.execute_on(node, Replicator.COMMANDS.RETRIEVE, timeout=RETRIEVE_TIMEOUT)
+        except Exception as e:
+            logger.warning(f"Could not retrieve the replica from {node.ip}:{node.port}: {e}")
+            continue
+
+        encoded = reply_result(response)
+        if not isinstance(encoded, str):
+            error = response.get("error") if isinstance(response, dict) else "bad reply"
+            logger.warning(f"{node.ip}:{node.port} could not send its replica: {error}")
+            continue
+
+        try:
+            return base64.b64decode(encoded, validate=True)
+        except ValueError:
+            logger.warning(f"{node.ip}:{node.port} sent a replica that is not valid base64")
+    return None
 
 
 class Replicator(Role):
@@ -23,43 +88,74 @@ class Replicator(Role):
         }
 
         self.storage = storage or ReplicaStorage()
+        self._write_lock = threading.Lock()
+        self._syncing = False
+        self._written_during_sync = {}
 
         super().__init__(commands, context)
 
     def start(self):
-        replicator_checksums = dict()
+        with self._start_lock:
+            if self._thread is not None:
+                return
+            self._thread = threading.Thread(target=self._sync_safely, daemon=True, name="ReplicaSync")
+            self._thread.start()
 
-        for node in self.context.nodes_with_role(Replicator):
-            if (node.ip, node.port) == self.context.membership().key:
-                continue
+    def _sync_safely(self):
+        try:
+            self.sync()
+        except Exception:
+            self.logger.exception("Replica sync failed")
 
-            response = self.context.execute_on(node, Replicator.COMMANDS.HASH_SUM)
-            if response['status'] != 'ok':
-                continue
+    def sync(self):
+        with self._write_lock:
+            self._syncing = True
+            self._written_during_sync = {}
+        try:
+            self._sync_from_majority()
+        finally:
+            with self._write_lock:
+                self._syncing = False
+                self._written_during_sync = {}
 
-            checksum = response['result']
+    def _sync_from_majority(self):
+        checksum, nodes = majority_copy(self.context)
+        if checksum is None or self._stop_event.is_set():
+            return
+        if checksum == self.hash_sum():
+            return
 
-            replicator_checksums[node] = checksum
+        data = fetch_copy(self.context, nodes)
+        if data is None or self._stop_event.is_set():
+            return
+        if not self.storage.archive_files(data) and self.storage.archive_files(self.storage.archive()):
+            self.logger.warning("The majority replica is empty, keeping the local files")
+            return
 
-        if replicator_checksums:
-            checksum = Counter(replicator_checksums.values()).most_common(1)[0][0]
-            matching_nodes = [
-                node
-                for node, node_checksum in replicator_checksums.items()
-                if node_checksum == checksum
-            ]
-
-            if checksum != self.hash_sum():
-                node = random.choice(matching_nodes)
-                response = self.context.execute_on(node, Replicator.COMMANDS.RETRIEVE)
-                self.storage.restore(base64.b64decode(response['result']))
+        with self._write_lock:
+            if self._stop_event.is_set():
+                return
+            self.storage.restore(data)
+            for filename, content in self._written_during_sync.items():
+                self.storage.save(filename, content)
+        self.logger.info("Replica synced from the cluster")
 
     def replicate(self, filename, content):
-        self.storage.save(filename, content)
+        with self._write_lock:
+            self.storage.save(filename, content)
+            if self._syncing:
+                self._written_during_sync[filename] = content
         return True
 
     def retrieve(self):
-        return base64.b64encode(self.storage.archive()).decode()
+        encoded = base64.b64encode(self.storage.archive()).decode()
+        if len(encoded) + FRAME_HEADROOM >= MAX_FRAME_SIZE:
+            self.logger.error(
+                f"The replica is {len(encoded)} bytes encoded, over the {MAX_FRAME_SIZE} byte frame limit. "
+                f"Relayed requests are also limited by gateway.max_payload_bytes"
+            )
+            raise ValueError("replica too large to retrieve")
+        return encoded
 
     def hash_sum(self):
         return self.storage.dir_checksum()

colony/core/roles/unit/role.py

@@ -60,3 +60,6 @@ class Unit(Role):
 
     def request_role(self, role):
         return self.context.get_role_manager().request_role(role)
+
+    def add_role_locally(self, role):
+        return self.context.get_role_manager().request_role(role, allow_on_demand=True)

colony/core/roles/vaultwarden/role.py

@@ -1,12 +1,15 @@
+import threading
 from enum import Enum
 
 from colony.core.storages.default import DefaultStorage
-from colony.core.roles.replicator.role import Replicator
+from colony.core.roles.replicator.role import Replicator, majority_copy, fetch_copy
 from colony.core.roles.role import Role, Command
 
 
 class VaultWarden(Role):
     REPLICATE_TIMEOUT = 3
+    RESTORE_ATTEMPTS = 5
+    RESTORE_RETRY_SECONDS = 5
 
     class COMMANDS(Enum):
         LS = 'VaultWarden.ls'
@@ -21,9 +24,18 @@ class VaultWarden(Role):
         }
 
         self.storage = DefaultStorage()
+        self._write_lock = threading.Lock()
+        self._written = False
 
         super().__init__(commands, context)
 
+    def start(self):
+        with self._start_lock:
+            if self._thread is not None:
+                return
+            self._thread = threading.Thread(target=self._restore_safely, daemon=True, name="VaultRestore")
+            self._thread.start()
+
     def ls(self, dir):
         return self.storage.ls(dir)
 
@@ -31,7 +43,9 @@ class VaultWarden(Role):
         return self.storage.open(file)
 
     def overwrite(self, filename, content):
-        path = self.storage.save(filename, content)
+        with self._write_lock:
+            path = self.storage.save(filename, content)
+            self._written = True
         if self.storage.SHARED:
             self._push_to_replicators(filename, content)
         return path
@@ -52,3 +66,57 @@ class VaultWarden(Role):
             if not isinstance(resp, dict) or resp.get("status") != "ok":
                 error = resp.get("error") if isinstance(resp, dict) else "bad reply"
                 self.logger.warning(f"{target} did not accept {filename}: {error}")
+
+    def _restore_safely(self):
+        try:
+            self.restore_from_replicas()
+        except Exception:
+            self.logger.exception("Could not restore the vault from the replicas")
+
+    def _local_replica(self):
+        replicator = self.context.self().get_role_instances().get(Replicator.__name__)
+        if replicator is None:
+            return None
+        return replicator.storage
+
+    def restore_from_replicas(self):
+        for attempt in range(self.RESTORE_ATTEMPTS):
+            if attempt > 0 and self._stop_event.wait(self.RESTORE_RETRY_SECONDS):
+                return
+            if self._restore_once():
+                return
+        self.logger.warning("No replica could be read, the vault starts from its local files")
+
+    def _restore_once(self):
+        local = self._local_replica()
+        local_checksum = None
+        if local is not None:
+            local_checksum = local.dir_checksum()
+
+        checksum, nodes = majority_copy(self.context, local_checksum=local_checksum)
+        if checksum is None:
+            return False
+
+        if checksum == local_checksum:
+            data = local.archive()
+        else:
+            data = fetch_copy(self.context, nodes)
+        if data is None:
+            return False
+
+        if not self.storage.archive_files(data):
+            self.logger.info("The replicas are empty, keeping the local vault files")
+            return True
+        if self.storage.dir_checksum() == checksum:
+            return True
+
+        with self._write_lock:
+            if self._stop_event.is_set():
+                return True
+            if self._written:
+                self.logger.info("The vault was written before the restore finished, keeping the local files")
+                return True
+            self.storage.restore(data)
+        self.logger.info("Restored the vault from the replicas")
+        return True
+

colony/core/storages/core.py

@@ -96,6 +96,38 @@ class Storage:
 
         return buffer.getvalue()
 
-    def restore(self, data):
+    @staticmethod
+    def archive_files(data):
+        names = set()
         with tarfile.open(fileobj=io.BytesIO(data), mode="r:gz") as tar:
-            tar.extractall(self.storage_dir, filter="data")
+            for member in tar.getmembers():
+                if member.isfile():
+                    names.add(member.name)
+        return names
+
+    def restore(self, data):
+        wanted = self.archive_files(data)
+        with _write_lock:
+            with tarfile.open(fileobj=io.BytesIO(data), mode="r:gz") as tar:
+                tar.extractall(self.storage_dir, filter="data")
+            root = Path(self.storage_dir)
+            self._prune_missing(root, wanted)
+            self._remove_empty_dirs(root)
+
+    @staticmethod
+    def _prune_missing(root, wanted):
+        for file in root.rglob("*"):
+            if not file.is_file() or file.name.startswith(".tmp-"):
+                continue
+            if file.relative_to(root).as_posix() not in wanted:
+                file.unlink()
+
+    @staticmethod
+    def _remove_empty_dirs(root):
+        directories = [p for p in root.rglob("*") if p.is_dir()]
+        directories.sort(key=lambda p: len(p.parts), reverse=True)
+        for directory in directories:
+            try:
+                directory.rmdir()
+            except OSError:
+                pass

colony/role_registry.py

@@ -17,11 +17,6 @@ class HardwareType(Enum):
     LONGEVITY = 5
 
 
-class HardwarePriority:
-    hardware: HardwareType
-    priority: int
-
-
 @dataclass
 class HardwarePreferences:
     priorities: dict[HardwareType, int]

colony/transport/whisperer.py

@@ -1,10 +1,14 @@
 import json
 import socket
 import ssl
+import threading
 
 from colony.settings import configuration
 from colony.transport.framing import send_frame, recv_frame
 
+_contexts = {}
+_contexts_lock = threading.Lock()
+
 
 class ConnectFailed(OSError):
     pass
@@ -19,6 +23,15 @@ class Whisperer:
         self.cert_file = cert_file or getattr(configuration.whisperer, "cert_file", "certs/server.crt")
         self.keyfile = keyfile or getattr(configuration.whisperer, "keyfile", "certs/server.key")
 
+    def _ssl_context(self):
+        key = (str(self.cert_file), str(self.keyfile))
+        with _contexts_lock:
+            context = _contexts.get(key)
+            if context is None:
+                context = self._build_ssl_context()
+                _contexts[key] = context
+            return context
+
     def _build_ssl_context(self):
         context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
         context.minimum_version = ssl.TLSVersion.TLSv1_2
@@ -30,7 +43,7 @@ class Whisperer:
 
     def send(self, payload, timeout=DEFAULT_TIMEOUT, connect_timeout=None):
         data = json.dumps(payload).encode()
-        context = self._build_ssl_context()
+        context = self._ssl_context()
 
         try:
             sock = socket.create_connection((self.host, self.port), timeout=connect_timeout or timeout)

colony/utils/network.py

@@ -4,9 +4,6 @@ import socket
 MAX_RANGE_SIZE = 1024
 
 
-def get_local_subnet():
-    return ipaddress.ip_network(f"{get_local_ip()}/24", strict=False)
-
 def get_local_ip():
     try:
         with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:

docker-compose.yml

@@ -54,11 +54,11 @@ services:
       COLONY_GATEWAY__ALLOW_COMMANDS: >-
         ["Gossip.Hello","Unit.ShowStats","Unit.ShowRoles","Unit.ShowHardware",
         "Unit.ShowCluster","Unit.RequestRole","Replicator.HashSum",
-        "Replicator.Retrieve","Replicator.Replicate"]
+        "Replicator.Retrieve","Replicator.Replicate","Chatter.Receive"]
 
       COLONY_ROLES__ON_DEMAND: "[]"
 
-      COLONY_ROLE_POLICY__MAX_ROLES: null
+      COLONY_ROLE_POLICY__MAX_ROLES: "@none"
       COLONY_ROLE_POLICY__DENY: "[]"
       COLONY_ROLE_POLICY__MIN_SCORES: "{}"
       COLONY_ROLE_POLICY__DECLINE_COOLDOWN: 60

pyproject.toml

@@ -4,7 +4,6 @@ version = "0.1.0"
 description = "Colony"
 requires-python = ">=3.14"
 dependencies = [
-    "dependency-injector==4.48.3",
     "dynaconf==3.2.12",
     "Flask==3.1.3",
     "GPUtil==1.4.0",

uv.lock

@@ -25,7 +25,6 @@ name = "colony"
 version = "0.1.0"
 source = { virtual = "." }
 dependencies = [
-    { name = "dependency-injector" },
     { name = "dynaconf" },
     { name = "flask" },
     { name = "gputil" },
@@ -35,7 +34,6 @@ dependencies = [
 
 [package.metadata]
 requires-dist = [
-    { name = "dependency-injector", specifier = "==4.48.3" },
     { name = "dynaconf", specifier = "==3.2.12" },
     { name = "flask", specifier = "==3.1.3" },
     { name = "gputil", specifier = "==1.4.0" },
@@ -43,21 +41,6 @@ requires-dist = [
     { name = "setuptools", specifier = "==80.9.0" },
 ]
 
-[[package]]
-name = "dependency-injector"
-version = "4.48.3"
-source = { registry = "https://pypi.org/simple" }
-sdist = { url = "https://files.pythonhosted.org/packages/9d/14/4f7b2d24d3ad50ac399262636cdf4b66b9318bfd0e9de5c61669625dc77c/dependency_injector-4.48.3.tar.gz", hash = "sha256:106e43a6a9959b2c49926edc45bdf07fdc377bc5fa1fd59012741590df30b20c", size = 1129251, upload-time = "2025-12-04T18:43:55.751Z" }
-wheels = [
-    { url = "https://files.pythonhosted.org/packages/b8/d1/bebb2921e1cfdeba992e57c62bcd871f4703fde852b1d64ceb0dd65540ee/dependency_injector-4.48.3-cp310-abi3-macosx_11_0_arm64.whl", hash = "sha256:37d5b962d9ba67f6536557dec3c83136cefd7a747bbe8ba53b158e34e050bae5", size = 1746065, upload-time = "2025-12-04T18:43:06.655Z" },
-    { url = "https://files.pythonhosted.org/packages/c3/da/068f23d12b55cda6e2d5e7783d1b602aa40f0889f35cf182ecfa1b12f7b8/dependency_injector-4.48.3-cp310-abi3-manylinux1_x86_64.manylinux_2_28_x86_64.manylinux_2_5_x86_64.whl", hash = "sha256:54aa2b8b8ff057555ab03ee98035eb41120d30f8ec220cf43aa3b117f6b9b1b1", size = 1829780, upload-time = "2025-12-04T18:43:08.497Z" },
-    { url = "https://files.pythonhosted.org/packages/e4/7f/7d5f16a87d6cc4111be41c8593ae6b98ebcdf0dc797c6e08e303ce6155b7/dependency_injector-4.48.3-cp310-abi3-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:6a1f8d5b7dcf24773c27ddb9beab043753d8e251804f2398468944b93c767a35", size = 1741838, upload-time = "2025-12-04T18:43:10.072Z" },
-    { url = "https://files.pythonhosted.org/packages/ea/1d/251c79858c9ebf53c6d8b51c9a653c794db0392070ab5a8f3f4a298a8b26/dependency_injector-4.48.3-cp310-abi3-musllinux_1_2_aarch64.whl", hash = "sha256:7ed11a944f5c84a5ad3f15bc9154f07affae2902967b4e47c74504d4595ed92f", size = 1726565, upload-time = "2025-12-04T18:43:11.892Z" },
-    { url = "https://files.pythonhosted.org/packages/45/92/023bd0b72baf170368323412a911fe31fd1c5cccbd4d399744bdec21cecd/dependency_injector-4.48.3-cp310-abi3-musllinux_1_2_x86_64.whl", hash = "sha256:00c7ad5b5b5f9b1616c363770eed8c629a42dd0155b398c4458c949b8c0b9ae0", size = 1820764, upload-time = "2025-12-04T18:43:13.765Z" },
-    { url = "https://files.pythonhosted.org/packages/ef/ae/0f6c65c6f67be9a7bcc18546ad08610b1cb294cef0da3d16502a1b336246/dependency_injector-4.48.3-cp310-abi3-win32.whl", hash = "sha256:2870a3b84ddb2e4b13731b7250b7714f0d4a07f6147a181b3ecbceed061b74f2", size = 1527160, upload-time = "2025-12-04T18:43:15.534Z" },
-    { url = "https://files.pythonhosted.org/packages/ae/74/51e8d051d55ae3b88fb0d94356cc9998e070a8b0cf6c93ef8456b8d05c01/dependency_injector-4.48.3-cp310-abi3-win_amd64.whl", hash = "sha256:d0e6244b3674a48ee2c5b0af29bf20d2b0b5baef806dd1739af9aab9d20f08de", size = 1655409, upload-time = "2025-12-04T18:43:16.901Z" },
-]
-
 [[package]]
 name = "dynaconf"
 version = "3.2.12"