diff --git a/repeater/data_acquisition/sqlite_handler.py b/repeater/data_acquisition/sqlite_handler.py
index d41cabd..ef088bf 100644
--- a/repeater/data_acquisition/sqlite_handler.py
+++ b/repeater/data_acquisition/sqlite_handler.py
@@ -30,6 +30,10 @@ class SQLiteHandler:
         # every write, which would defeat the cache under load.
         self._cumulative_counts_cache = {"timestamp": 0.0, "value": None}
         self._cumulative_counts_ttl_sec = 3.0
+        # Optional callback fired after any transport_keys (region) write, so the
+        # daemon can rebuild the flood-reply RegionMap. Fired after commit, in the
+        # writer's thread; see set_transport_keys_changed_callback.
+        self._transport_keys_changed_cb = None
         # Thread-local storage for persistent SQLite connections.
         # Opening a new connection on every DB call is expensive on SD-card
         # storage: each sqlite3.connect() call triggers file-system operations
@@ -80,6 +84,26 @@ class SQLiteHandler:
         self._packet_stats_cache.clear()
         self._neighbors_cache = {"timestamp": 0.0, "value": None}
 
+    def set_transport_keys_changed_callback(self, callback) -> None:
+        """Register a callback fired after any transport_keys (region) write.
+
+        The daemon uses this to rebuild the flood-reply RegionMap when a named
+        region is added, removed, or has its flood policy changed (CLI, web API,
+        or Glass sync all funnel through the create/update/delete/sync methods
+        below). The callback runs after the write commits, in the writer's
+        thread; pass ``None`` to clear it.
+        """
+        self._transport_keys_changed_cb = callback
+
+    def _notify_transport_keys_changed(self) -> None:
+        callback = self._transport_keys_changed_cb
+        if callback is None:
+            return
+        try:
+            callback()
+        except Exception as e:
+            logger.error(f"transport_keys change callback failed: {e}", exc_info=True)
+
     def _init_database(self):
         try:
             with self._connect() as conn:
@@ -2798,7 +2822,9 @@ class SQLiteHandler:
                         current_time,
                     ),
                 )
-                return cursor.lastrowid
+                new_id = cursor.lastrowid
+            self._notify_transport_keys_changed()
+            return new_id
         except Exception as e:
             logger.error(f"Failed to create transport key: {e}")
             return None
@@ -2917,7 +2943,10 @@ class SQLiteHandler:
                     """,
                     params,
                 )
-                return cursor.rowcount > 0
+                changed = cursor.rowcount > 0
+            if changed:
+                self._notify_transport_keys_changed()
+            return changed
         except Exception as e:
             logger.error(f"Failed to update transport key: {e}")
             return False
@@ -2926,7 +2955,10 @@ class SQLiteHandler:
         try:
             with self._connect() as conn:
                 cursor = conn.execute("DELETE FROM transport_keys WHERE id = ?", (key_id,))
-                return cursor.rowcount > 0
+                changed = cursor.rowcount > 0
+            if changed:
+                self._notify_transport_keys_changed()
+            return changed
         except Exception as e:
             logger.error(f"Failed to delete transport key: {e}")
             return False
@@ -3043,6 +3075,7 @@ class SQLiteHandler:
                 db_ids[node["node_id"]] = int(cursor.lastrowid)
             conn.commit()
 
+        self._notify_transport_keys_changed()
         return {"applied_nodes": len(ordered), "generated_keys": generated_keys}
 
     def delete_advert(self, advert_id: int) -> bool:
diff --git a/repeater/identity_manager.py b/repeater/identity_manager.py
index 788a41e..28ea4c7 100644
--- a/repeater/identity_manager.py
+++ b/repeater/identity_manager.py
@@ -5,6 +5,23 @@ from typing import Any, Dict, Iterable, Optional, Tuple
 logger = logging.getLogger("IdentityManager")
 
 
+def _namespace_for(identity_type: str) -> str:
+    """Routing/persistence namespace a local identity occupies.
+
+    A one-byte dest-hash collision is only unrepresentable when both
+    identities share a namespace.  Companions key their runtime bridge
+    (``daemon.companion_bridges[hash]``) and their SQLite state
+    (``companion_prefs`` PRIMARY KEY, plus the contacts/channels/messages
+    tables) by the hash byte.  The repeater and every room server instead
+    share the login/text/protocol helper ``handlers[hash]`` slot and the
+    ``room_*`` tables, so they form one "server" namespace.  A companion and
+    a server-side identity live in physically separate stores, and the packet
+    router (``_consume_via_local_candidates``) offers a colliding packet to
+    both and lets HMAC pick the owner, so they may safely share a prefix.
+    """
+    return "companion" if identity_type == "companion" else "server"
+
+
 class IdentityConfigurationError(RuntimeError):
     """A configured local identity cannot be represented safely."""
 
@@ -30,25 +47,30 @@ class IdentitySpec:
 class IdentityManager:
     def __init__(self, config: dict):
         self.config = config
-        self.identities: Dict[int, Tuple[Any, dict, str]] = {}
+        self.identities: Dict[Tuple[int, str], Tuple[Any, dict, str]] = {}
         self.named_identities: Dict[str, Tuple[Any, dict, str]] = {}
-        self.registered_hashes: Dict[int, str] = {}
+        self.registered_hashes: Dict[Tuple[int, str], str] = {}
 
-    def registration_error(self, name: str, identity) -> Optional[str]:
+    def registration_error(self, name: str, identity, identity_type: str) -> Optional[str]:
         """Return a reason this identity cannot be registered, or ``None``.
 
-        Local protocol routing and companion persistence are keyed by the first
-        byte of the public key.  A collision therefore cannot be represented
-        safely, even though the full public keys differ.  Names must also be
-        unique because callers use them to locate the configured service.
+        Local protocol routing and per-identity persistence are keyed by the
+        first public-key byte *within a namespace* (see :func:`_namespace_for`).
+        Two identities that share both the hash byte and the namespace cannot
+        be represented safely even though their full keys differ; a companion
+        and a server-side identity may share a prefix because they live in
+        separate stores that the packet router disambiguates by HMAC.  Names
+        must also be unique because callers use them to locate the service.
         """
         hash_byte = identity.get_public_key()[0]
+        key = (hash_byte, _namespace_for(identity_type))
 
-        if hash_byte in self.identities:
-            existing_name = self.registered_hashes.get(hash_byte, "unknown")
+        if key in self.identities:
+            existing_name = self.registered_hashes.get(key, "unknown")
             return (
                 f"Identity '{name}' (hash=0x{hash_byte:02X}) conflicts with "
-                f"existing identity '{existing_name}'"
+                f"existing identity '{existing_name}'; identities of the same "
+                f"class must have unique one-byte public-key prefixes"
             )
 
         if name in self.named_identities:
@@ -66,17 +88,17 @@ class IdentityManager:
 
         Each spec is checked against the currently registered identities (via
         :meth:`registration_error`) and against the other specs in the batch,
-        without mutating any state. Local protocol routing and companion
-        persistence are keyed by the first public-key byte, so a one-byte
-        prefix collision cannot be represented even though the full keys
-        differ, and names must be unique because callers use them to locate
-        the configured service.
+        without mutating any state. A one-byte prefix collision is only
+        rejected when both identities share a namespace (see
+        :func:`_namespace_for`); a companion may share a prefix with a
+        server-side identity. Names must be unique because callers use them to
+        locate the configured service.
         """
-        batch_hashes: Dict[int, IdentitySpec] = {}
+        batch_keys: Dict[Tuple[int, str], IdentitySpec] = {}
         batch_names: Dict[str, IdentitySpec] = {}
 
         for spec in specs:
-            error = self.registration_error(spec.name, spec.identity)
+            error = self.registration_error(spec.name, spec.identity, spec.identity_type)
             if error:
                 raise IdentityConfigurationError(error)
             existing = batch_names.get(spec.name)
@@ -85,52 +107,69 @@ class IdentityManager:
                     f"Local identity name '{spec.name}' conflicts with existing "
                     f"identity '{existing.label}'"
                 )
-            existing = batch_hashes.get(spec.hash_byte)
+            key = (spec.hash_byte, _namespace_for(spec.identity_type))
+            existing = batch_keys.get(key)
             if existing is not None:
                 raise IdentityConfigurationError(
                     f"Local identity '{spec.label}' (hash=0x{spec.hash_byte:02X}) "
-                    f"conflicts with '{existing.label}'; local identities must "
-                    "have unique one-byte public-key prefixes"
+                    f"conflicts with '{existing.label}'; identities of the same "
+                    "class must have unique one-byte public-key prefixes"
                 )
             batch_names[spec.name] = spec
-            batch_hashes[spec.hash_byte] = spec
+            batch_keys[key] = spec
 
-    def validate_identity(self, name: str, identity) -> bool:
+    def validate_identity(self, name: str, identity, identity_type: str) -> bool:
         """Log and report whether an identity can be registered without mutation."""
-        error = self.registration_error(name, identity)
+        error = self.registration_error(name, identity, identity_type)
         if error:
             logger.error("Identity registration rejected: %s", error)
             return False
         return True
 
     def register_identity(self, name: str, identity, config: dict, identity_type: str):
-        if not self.validate_identity(name, identity):
+        if not self.validate_identity(name, identity, identity_type):
             return False
 
         hash_byte = identity.get_public_key()[0]
+        key = (hash_byte, _namespace_for(identity_type))
 
-        self.identities[hash_byte] = (identity, config, identity_type)
+        self.identities[key] = (identity, config, identity_type)
         self.named_identities[name] = (identity, config, identity_type)
-        self.registered_hashes[hash_byte] = f"{identity_type}:{name}"
+        self.registered_hashes[key] = f"{identity_type}:{name}"
 
         logger.info(
             f"Identity registered: name={name}, hash=0x{hash_byte:02X}, type={identity_type}"
         )
         return True
 
-    def get_identity_by_hash(self, hash_byte: int) -> Optional[Tuple[Any, dict, str]]:
-        return self.identities.get(hash_byte)
+    def get_identity_by_hash(
+        self, hash_byte: int, namespace: Optional[str] = None
+    ) -> Optional[Tuple[Any, dict, str]]:
+        """Return the identity registered at ``hash_byte``.
+
+        With ``namespace`` given, returns that namespace's identity; otherwise
+        returns the first identity at the hash (a companion and a server-side
+        identity may both be registered there since they no longer collide).
+        """
+        if namespace is not None:
+            return self.identities.get((hash_byte, namespace))
+        for (registered_hash, _ns), value in self.identities.items():
+            if registered_hash == hash_byte:
+                return value
+        return None
 
     def get_identity_by_name(self, name: str) -> Optional[Tuple[Any, dict, str]]:
         return self.named_identities.get(name)
 
-    def has_identity(self, hash_byte: int) -> bool:
-        return hash_byte in self.identities
+    def has_identity(self, hash_byte: int, namespace: Optional[str] = None) -> bool:
+        if namespace is not None:
+            return (hash_byte, namespace) in self.identities
+        return any(registered_hash == hash_byte for registered_hash, _ns in self.identities)
 
     def list_identities(self) -> list:
         identities = []
-        for hash_byte, (identity, config, id_type) in self.identities.items():
-            name = self.registered_hashes.get(hash_byte, "unknown")
+        for (hash_byte, namespace), (identity, config, id_type) in self.identities.items():
+            name = self.registered_hashes.get((hash_byte, namespace), "unknown")
             identities.append(
                 {
                     "hash": f"0x{hash_byte:02X}",
diff --git a/repeater/main.py b/repeater/main.py
index 468ddd4..aa3f2b0 100644
--- a/repeater/main.py
+++ b/repeater/main.py
@@ -37,6 +37,7 @@ from repeater.handler_helpers import (
 from repeater.identity_manager import IdentityConfigurationError, IdentityManager, IdentitySpec
 from repeater.logging_utils import normalize_log_level
 from repeater.packet_router import PacketRouter
+from repeater.region_map_builder import build_region_map
 from repeater.sensors import SensorManager
 from repeater.utils_packet import create_scoped_advert_packet
 from repeater.web.http_server import HTTPStatsServer, _log_buffer
@@ -107,6 +108,11 @@ class RepeaterDaemon:
         self.router = None
         self.companion_bridges: dict[int, object] = {}
         self.companion_frame_servers: list = []
+        # Shared RegionMap describing the named regions this repeater serves.
+        # Wired into the dispatcher and every companion bridge so core can
+        # re-scope flood replies to the region their request arrived under
+        # (firmware sendFloodReply parity). Rebuilt on any transport_keys change.
+        self._region_map = None
         # Parsed once during the startup preflight; the identity loaders reuse
         # them so config parsing (and its warnings) does not run twice.
         self._room_server_specs: list[IdentitySpec] | None = None
@@ -213,6 +219,56 @@ class RepeaterDaemon:
         manager = self.identity_manager or IdentityManager(self.config)
         manager.validate_specs(specs)
 
+    def _get_sqlite_handler(self):
+        """Return the shared SQLiteHandler, or None if storage is unavailable."""
+        handler = self.repeater_handler
+        storage = getattr(handler, "storage", None) if handler else None
+        return getattr(storage, "sqlite_handler", None) if storage else None
+
+    def _init_region_map(self) -> None:
+        """Build the shared RegionMap, wire it into the dispatcher, and hook rebuilds.
+
+        Called once storage is available (right after ``repeater_handler`` is
+        created) and before any companion bridge is built, so the dispatcher and
+        every bridge share the same instance. Runtime region edits (transport_keys
+        CRUD from the CLI, web API, or Glass sync) fire the storage change hook,
+        which reruns ``refresh_region_map``.
+        """
+        sqlite_handler = self._get_sqlite_handler()
+        self._region_map = build_region_map(self.config, sqlite_handler)
+        if self.dispatcher is not None:
+            self.dispatcher.region_map = self._region_map
+        if sqlite_handler is not None and hasattr(
+            sqlite_handler, "set_transport_keys_changed_callback"
+        ):
+            sqlite_handler.set_transport_keys_changed_callback(self.refresh_region_map)
+        logger.info(
+            "Region map initialized with %d served region(s)",
+            len(self._region_map.regions),
+        )
+
+    def refresh_region_map(self) -> None:
+        """Rebuild the RegionMap and reassign it to the dispatcher and all bridges.
+
+        Fires from the storage transport_keys change hook whenever a named region
+        is added, removed, or has its flood policy changed. A fresh instance is
+        reassigned (rather than mutated in place) because this may run in a
+        cherrypy worker thread while ``find_match`` iterates the map on the RX
+        hot path in the event-loop thread: an attribute rebind is atomic under
+        the GIL, so an in-flight match keeps using the old, fully-built map.
+        New bridges pick up the current instance at creation time.
+        """
+        new_map = build_region_map(self.config, self._get_sqlite_handler())
+        self._region_map = new_map
+        if self.dispatcher is not None:
+            self.dispatcher.region_map = new_map
+        for bridge in list(self.companion_bridges.values()):
+            try:
+                bridge.region_map = new_map
+            except Exception:
+                logger.debug("Failed to update region map on a companion bridge", exc_info=True)
+        logger.info("Region map refreshed with %d served region(s)", len(new_map.regions))
+
     async def initialize(self):
 
         logger.info(f"Initializing repeater: {self.config['repeater']['node_name']}")
@@ -386,6 +442,11 @@ class RepeaterDaemon:
                 send_advert_func=self.send_advert,
             )
 
+            # Storage now exists: build the served-region map and wire it into the
+            # dispatcher so flood replies are re-scoped to their request's region.
+            # Runs before any companion bridge is created so all share one instance.
+            self._init_region_map()
+
             # Create router
             self.router = PacketRouter(self)
             await self.router.start()
@@ -784,6 +845,10 @@ class RepeaterDaemon:
                     **bridge_kwargs,
                 )
 
+                # Share the dispatcher's served-region map so this bridge re-scopes
+                # its own flood replies to the region the request arrived under.
+                bridge.region_map = self._region_map
+
                 # Restore persisted state (contacts/channels/messages) from SQLite.
                 # Raises CompanionStateLoadError instead of continuing with an
                 # empty store when persisted rows exist but cannot be loaded.
@@ -962,7 +1027,7 @@ class RepeaterDaemon:
 
         if self.identity_manager is None:
             raise RuntimeError("Identity manager must be initialized before adding a companion")
-        registration_error = self.identity_manager.registration_error(name, identity)
+        registration_error = self.identity_manager.registration_error(name, identity, "companion")
         if registration_error:
             raise ValueError(f"Cannot add companion: {registration_error}")
 
@@ -1018,6 +1083,10 @@ class RepeaterDaemon:
             **bridge_kwargs,
         )
 
+        # Share the current served-region map (hot-reload path) so this bridge
+        # re-scopes its flood replies to the region the request arrived under.
+        bridge.region_map = self._region_map
+
         # Restore persisted state; raises CompanionStateLoadError when persisted
         # rows exist but cannot be loaded (hot-reload callers surface the error).
         if sqlite_handler:
@@ -1704,6 +1773,11 @@ def main():
         asyncio.run(daemon.run())
     except KeyboardInterrupt:
         logger.info("Repeater stopped")
+    except IdentityConfigurationError as e:
+        # A misconfigured local identity is an actionable config problem, not a
+        # crash: report just the message so the fix is obvious, no stack trace.
+        logger.error("Identity configuration error: %s", e)
+        sys.exit(1)
     except Exception as e:
         logger.error(f"Fatal error: {e}", exc_info=True)
         sys.exit(1)
diff --git a/repeater/region_map_builder.py b/repeater/region_map_builder.py
new file mode 100644
index 0000000..052d329
--- /dev/null
+++ b/repeater/region_map_builder.py
@@ -0,0 +1,124 @@
+"""Build a core :class:`RegionMap` from the repeater's served transport-key regions.
+
+Core's flood-reply scoping (``region_map.apply_reply_scope``) re-scopes a flood
+reply to the region its request arrived under, mirroring firmware
+``simple_repeater::sendFloodReply``. For that to engage, the dispatcher and every
+companion bridge need a ``RegionMap`` describing the named regions this repeater
+serves. This module builds that map from the ``transport_keys`` table — the same
+source ``login.LoginHelper._format_region_names`` reads.
+
+Firmware-parity notes:
+
+- The ``*`` wildcard (unscoped flood) is deliberately **not** a region entry. A
+  plain FLOOD request replies plain, so ``find_match`` must return ``None`` for
+  it. Wildcard handling lives in ``capture_recv_region`` (route-type based), not
+  here, and ``mesh.unscoped_flood_allow`` never changes the map contents.
+- A deny-flood region carries ``REGION_DENY_FLOOD`` so
+  ``find_match(mask=REGION_DENY_FLOOD)`` skips it => its request replies plain.
+- The transport key is derived from the region name via ``get_auto_key_for`` —
+  the same derivation senders and the repeater's own outgoing floods use. An
+  explicit stored key is only carried when it is genuinely custom material the
+  name would not reproduce (a private ``$`` region, or imported key material via
+  Glass sync); a redundant key that disagreed with the name would silently
+  re-scope replies to the wrong code.
+"""
+
+from __future__ import annotations
+
+import base64
+import binascii
+import logging
+from typing import Optional
+
+from openhop_core.protocol.region_map import REGION_DENY_FLOOD, RegionEntry, RegionMap
+from openhop_core.protocol.transport_keys import get_auto_key_for
+
+logger = logging.getLogger("RepeaterRegionMap")
+
+
+def _decode_stored_key(raw) -> Optional[bytes]:
+    """Decode a stored ``transport_key`` to 16 raw bytes, or ``None``.
+
+    Keys are stored base64-encoded (see ``SQLiteHandler.generate_transport_key``);
+    tolerate raw bytes too. Anything that is not exactly 16 bytes is ignored, so a
+    corrupt or wrong-length key falls back to name hashing rather than breaking
+    matching.
+    """
+    if not raw:
+        return None
+    if isinstance(raw, (bytes, bytearray)):
+        key = bytes(raw)
+    else:
+        try:
+            key = base64.b64decode(str(raw), validate=True)
+        except (binascii.Error, ValueError):
+            return None
+    return key if len(key) == 16 else None
+
+
+def build_region_map(config, sqlite_handler) -> RegionMap:
+    """Return a :class:`RegionMap` of the named regions this repeater serves.
+
+    ``config`` is currently unused (the ``*`` wildcard is not a map entry) but is
+    kept in the signature so a future config-driven region source stays a
+    drop-in change for every caller.
+    """
+    region_map = RegionMap()
+    if sqlite_handler is None:
+        return region_map
+
+    try:
+        records = sqlite_handler.get_transport_keys()
+    except Exception as exc:  # defensive: never let a bad read break startup
+        logger.warning("Failed to read transport keys for region map: %s", exc)
+        return region_map
+
+    for rec in records or []:
+        name = (rec.get("name") or "").strip()
+        # Skip empty names and the wildcard: a plain/unscoped flood replies plain,
+        # so find_match must not resolve it to a region.
+        if not name or name == "*":
+            continue
+
+        flood_policy = (rec.get("flood_policy") or "deny").strip().lower()
+        flags = 0 if flood_policy == "allow" else REGION_DENY_FLOOD
+
+        private_keys = None
+        key_bytes = _decode_stored_key(rec.get("transport_key"))
+        if key_bytes is not None:
+            if name.startswith("$"):
+                # Private region: core never name-hashes a "$" name, so the stored
+                # key is the only usable key. Without it the region matches nothing.
+                private_keys = [key_bytes]
+            else:
+                # Public region: rely on name hashing unless the stored key is
+                # genuinely custom material the name would not reproduce.
+                try:
+                    derived = get_auto_key_for(name)
+                except ValueError:
+                    derived = None
+                if derived != key_bytes:
+                    private_keys = [key_bytes]
+
+        try:
+            region_id = int(rec.get("id") or 0)
+        except (TypeError, ValueError):
+            region_id = 0
+
+        parent_raw = rec.get("parent_id")
+        try:
+            parent = int(parent_raw) if parent_raw is not None else 0
+        except (TypeError, ValueError):
+            parent = 0
+
+        region_map.add_region(
+            RegionEntry(
+                id=region_id,
+                parent=parent,
+                flags=flags,
+                name=name,
+                private_keys=private_keys,
+            )
+        )
+
+    return region_map
diff --git a/tests/test_identity_collision_preflight.py b/tests/test_identity_collision_preflight.py
index e0abdd2..3c1759f 100644
--- a/tests/test_identity_collision_preflight.py
+++ b/tests/test_identity_collision_preflight.py
@@ -21,21 +21,41 @@ class _SeedFirstByteIdentity:
         return self._public_key[:3]
 
 
-def _config(*, companions=()):
+def _config(*, companions=(), room_servers=()):
     return {
         "repeater": {"node_name": "n", "identity_key": b"\x10" * 32},
         "logging": {},
-        "identities": {"companions": list(companions)},
+        "identities": {
+            "companions": list(companions),
+            "room_servers": list(room_servers),
+        },
     }
 
 
-def test_startup_preflight_rejects_default_repeater_hash_collision():
+def test_startup_preflight_allows_companion_sharing_repeater_prefix():
+    """A companion may share the repeater's one-byte prefix: they live in
+    separate runtime stores (companion_bridges vs helper.handlers) and DB
+    tables, and the packet router disambiguates them on-air by HMAC."""
     daemon = RepeaterDaemon(
         _config(companions=({"name": "comp", "identity_key": "10" * 32},)),
         radio=object(),
     )
     local_identity = _SeedFirstByteIdentity(b"\x10" * 32)
 
+    with patch("openhop_core.LocalIdentity", _SeedFirstByteIdentity):
+        # Must not raise: the cross-namespace collision is representable.
+        daemon._preflight_configured_local_identities(local_identity)
+
+
+def test_startup_preflight_rejects_repeater_room_server_prefix_collision():
+    """The repeater and a room server share the server namespace (the login/
+    text helper handlers[hash] slot), so a prefix collision is still rejected."""
+    daemon = RepeaterDaemon(
+        _config(room_servers=({"name": "room", "identity_key": "10" * 32},)),
+        radio=object(),
+    )
+    local_identity = _SeedFirstByteIdentity(b"\x10" * 32)
+
     with patch("openhop_core.LocalIdentity", _SeedFirstByteIdentity):
         with pytest.raises(IdentityConfigurationError, match="one-byte public-key prefixes"):
             daemon._preflight_configured_local_identities(local_identity)
@@ -93,10 +113,12 @@ async def test_invalid_config_entry_logs_once_across_preflight_and_load(caplog):
 
 @pytest.mark.asyncio
 async def test_hot_added_companion_collision_is_rejected_before_stateful_setup():
+    # A second companion sharing an already-registered companion's prefix must
+    # be rejected (same companion namespace: companion_bridges + companion_* DB).
     daemon = RepeaterDaemon(_config(), radio=object())
     daemon.identity_manager = IdentityManager({})
     daemon.identity_manager.register_identity(
-        "repeater", _SeedFirstByteIdentity(b"\x33" * 32), {}, "repeater"
+        "existing", _SeedFirstByteIdentity(b"\x33" * 32), {}, "companion"
     )
 
     comp_config = {"name": "comp", "identity_key": "33" * 32, "settings": {}}
diff --git a/tests/test_identity_manager.py b/tests/test_identity_manager.py
index ead57fd..6ee3298 100644
--- a/tests/test_identity_manager.py
+++ b/tests/test_identity_manager.py
@@ -34,8 +34,8 @@ def test_identity_manager_rejects_duplicate_names_without_mutating_state():
     id_b = _FakeIdentity(bytes([0x22]) + b"B" * 31)
 
     assert mgr.register_identity("alpha", id_a, {}, "repeater") is True
-    assert "already registered" in mgr.registration_error("alpha", id_b)
-    assert mgr.validate_identity("alpha", id_b) is False
+    assert "already registered" in mgr.registration_error("alpha", id_b, "companion")
+    assert mgr.validate_identity("alpha", id_b, "companion") is False
     assert mgr.get_identity_by_hash(0x22) is None
 
 
@@ -63,15 +63,17 @@ def test_identity_manager_list_and_type_filtering():
 
 def test_identity_manager_list_handles_none_identity_fields():
     mgr = IdentityManager(config={})
-    mgr.identities[0x44] = (None, {}, "repeater")
-    mgr.registered_hashes[0x44] = "repeater:ghost"
+    mgr.identities[(0x44, "server")] = (None, {}, "repeater")
+    mgr.registered_hashes[(0x44, "server")] = "repeater:ghost"
 
     listed = mgr.list_identities()
     assert listed[0]["address"] == "N/A"
     assert listed[0]["public_key"] is None
 
 
-def test_validate_specs_rejects_intra_batch_hash_collision():
+def test_validate_specs_rejects_intra_batch_same_namespace_hash_collision():
+    """Two server-side identities (repeater + room server) share the login/
+    text helper handlers[hash] slot, so a prefix collision is unrepresentable."""
     mgr = IdentityManager(config={})
     id_a = _FakeIdentity(bytes([0x11]) + b"A" * 31)
     id_b = _FakeIdentity(bytes([0x11]) + b"B" * 31)
@@ -80,11 +82,63 @@ def test_validate_specs_rejects_intra_batch_hash_collision():
         mgr.validate_specs(
             [
                 IdentitySpec("alpha", id_a, {}, "repeater"),
+                IdentitySpec("beta", id_b, {}, "room_server"),
+            ]
+        )
+
+
+def test_validate_specs_rejects_two_companions_same_prefix():
+    """Two companions share companion_bridges[hash] and the companion_* DB
+    keying, so their prefix must be unique."""
+    mgr = IdentityManager(config={})
+    id_a = _FakeIdentity(bytes([0x11]) + b"A" * 31)
+    id_b = _FakeIdentity(bytes([0x11]) + b"B" * 31)
+
+    with pytest.raises(IdentityConfigurationError, match="one-byte public-key prefixes"):
+        mgr.validate_specs(
+            [
+                IdentitySpec("alpha", id_a, {}, "companion"),
                 IdentitySpec("beta", id_b, {}, "companion"),
             ]
         )
 
 
+def test_validate_specs_allows_companion_sharing_prefix_with_server_identity():
+    """A companion and a server-side identity live in separate runtime stores
+    and DB tables (the router disambiguates by HMAC), so they may share a
+    one-byte prefix. Verified for both the repeater and a room server."""
+    mgr = IdentityManager(config={})
+    repeater = _FakeIdentity(bytes([0x11]) + b"R" * 31)
+    room = _FakeIdentity(bytes([0x22]) + b"S" * 31)
+    comp_vs_repeater = _FakeIdentity(bytes([0x11]) + b"C" * 31)
+    comp_vs_room = _FakeIdentity(bytes([0x22]) + b"D" * 31)
+
+    mgr.validate_specs(
+        [
+            IdentitySpec("rep", repeater, {}, "repeater"),
+            IdentitySpec("room", room, {}, "room_server"),
+            IdentitySpec("comp-a", comp_vs_repeater, {}, "companion"),
+            IdentitySpec("comp-b", comp_vs_room, {}, "companion"),
+        ]
+    )
+
+
+def test_register_companion_sharing_repeater_prefix_keeps_both():
+    """Cross-namespace registration keeps both entries addressable."""
+    mgr = IdentityManager(config={})
+    repeater = _FakeIdentity(bytes([0x11]) + b"R" * 31)
+    companion = _FakeIdentity(bytes([0x11]) + b"C" * 31)
+
+    assert mgr.register_identity("rep", repeater, {}, "repeater") is True
+    assert mgr.register_identity("comp", companion, {}, "companion") is True
+
+    assert mgr.get_identity_by_hash(0x11, "server")[0] is repeater
+    assert mgr.get_identity_by_hash(0x11, "companion")[0] is companion
+    assert mgr.has_identity(0x11, "server") is True
+    assert mgr.has_identity(0x11, "companion") is True
+    assert len(mgr.list_identities()) == 2
+
+
 def test_validate_specs_rejects_intra_batch_duplicate_name():
     mgr = IdentityManager(config={})
     id_a = _FakeIdentity(bytes([0x11]) + b"A" * 31)
@@ -107,8 +161,9 @@ def test_validate_specs_rejects_registered_collisions_without_mutation():
 
     assert mgr.register_identity("alpha", id_a, {}, "repeater") is True
 
+    # A room server (server namespace) collides with the registered repeater.
     with pytest.raises(IdentityConfigurationError, match="conflicts"):
-        mgr.validate_specs([IdentitySpec("beta", id_hash_collision, {}, "companion")])
+        mgr.validate_specs([IdentitySpec("beta", id_hash_collision, {}, "room_server")])
     with pytest.raises(IdentityConfigurationError, match="already registered"):
         mgr.validate_specs([IdentitySpec("alpha", id_name_collision, {}, "companion")])
 
diff --git a/tests/test_main_py_coverage.py b/tests/test_main_py_coverage.py
index 94c1774..85f28e5 100644
--- a/tests/test_main_py_coverage.py
+++ b/tests/test_main_py_coverage.py
@@ -1,9 +1,11 @@
+import logging
 from types import SimpleNamespace
 from unittest.mock import AsyncMock, MagicMock, call, patch
 
 import pytest
 
 from repeater.companion.constants import STATS_TYPE_CORE, STATS_TYPE_PACKETS, STATS_TYPE_RADIO
+from repeater.identity_manager import IdentityConfigurationError
 from repeater.main import RepeaterDaemon
 from repeater.main import main as repeater_main
 from openhop_core.node.dispatcher import Dispatcher
@@ -533,3 +535,35 @@ def test_main_entrypoint_success_and_fatal_paths(monkeypatch):
             repeater_main()
 
     exit_mock.assert_called_once_with(1)
+
+
+def test_main_identity_config_error_exits_cleanly_without_traceback(caplog):
+    """A configured-identity collision exits 1 with a clean message and no
+    stack-trace dump (regression: the fatal handler used to log exc_info=True
+    for every exception)."""
+
+    class _Args:
+        config = "/tmp/test.yaml"
+        log_level = None
+
+    fake_daemon = SimpleNamespace(run=MagicMock(return_value=object()))
+    err = IdentityConfigurationError(
+        "Local identity 'companion:B' (hash=0x77) conflicts with 'companion:A'"
+    )
+
+    with (
+        patch("argparse.ArgumentParser.parse_args", return_value=_Args()),
+        patch("repeater.main.load_config", return_value=_base_config()),
+        patch("repeater.main.RepeaterDaemon", return_value=fake_daemon),
+        patch("asyncio.run", side_effect=err),
+        patch("sys.exit", side_effect=SystemExit(1)) as exit_mock,
+        caplog.at_level(logging.ERROR),
+    ):
+        with pytest.raises(SystemExit):
+            repeater_main()
+
+    exit_mock.assert_called_once_with(1)
+    config_errors = [r for r in caplog.records if "Identity configuration error" in r.getMessage()]
+    assert len(config_errors) == 1
+    assert config_errors[0].exc_info is None  # no traceback attached
+    assert not any("Fatal error" in r.getMessage() for r in caplog.records)
diff --git a/tests/test_region_map_wiring.py b/tests/test_region_map_wiring.py
new file mode 100644
index 0000000..6a9d1cd
--- /dev/null
+++ b/tests/test_region_map_wiring.py
@@ -0,0 +1,292 @@
+"""Served-region map: build correctness + dispatcher/bridge wiring.
+
+Core re-scopes a flood reply to the region its request arrived under
+(``region_map.apply_reply_scope``), but only when a ``RegionMap`` is wired onto
+the dispatcher and companion bridges. This repeater builds that map from the
+``transport_keys`` table (``build_region_map``), assigns one shared instance to
+the dispatcher and every bridge, and rebuilds it whenever a region is added,
+removed, or re-flooded via the storage change hook.
+
+These tests cover the repeater's contribution: the record -> RegionEntry mapping
+and flood matching, the storage change hook, and the daemon wiring that keeps a
+non-None map on the dispatcher and all live bridges after a runtime change.
+"""
+
+import base64
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+
+from openhop_core.protocol.packet import Packet
+from openhop_core.protocol.region_map import REGION_DENY_FLOOD
+from openhop_core.protocol.transport_keys import get_auto_key_for, scope_packet
+
+from repeater.data_acquisition.sqlite_handler import SQLiteHandler
+from repeater.main import RepeaterDaemon
+from repeater.region_map_builder import build_region_map
+
+
+class _FakeHandler:
+    def __init__(self, records):
+        self._records = records
+
+    def get_transport_keys(self):
+        return self._records
+
+
+def _b64key(name):
+    return base64.b64encode(get_auto_key_for(name)).decode("ascii")
+
+
+def _scoped_flood(key, payload=b"reply-body"):
+    """A TRANSPORT_FLOOD packet whose transport code was hashed with ``key``."""
+    pkt = Packet()
+    pkt.payload = bytearray(payload)
+    scope_packet(pkt, key)
+    return pkt
+
+
+def _plain_flood(payload=b"reply-body"):
+    pkt = Packet()
+    pkt.payload = bytearray(payload)
+    pkt.header = 0x01  # ROUTE_TYPE_FLOOD, no transport codes
+    return pkt
+
+
+# ---------------------------------------------------------------------------
+# build_region_map: record -> RegionEntry mapping and flood matching
+# ---------------------------------------------------------------------------
+def test_allow_region_matches_scoped_flood():
+    rm = build_region_map(
+        {},
+        _FakeHandler(
+            [
+                {
+                    "id": 1,
+                    "name": "#usa",
+                    "flood_policy": "allow",
+                    "transport_key": _b64key("#usa"),
+                    "parent_id": None,
+                }
+            ]
+        ),
+    )
+    match = rm.find_match(_scoped_flood(get_auto_key_for("#usa")), mask=REGION_DENY_FLOOD)
+    assert match is not None
+    assert match.name == "#usa"
+    assert match.flags == 0
+
+
+def test_deny_region_is_skipped_under_flood_mask():
+    rm = build_region_map(
+        {},
+        _FakeHandler(
+            [
+                {
+                    "id": 2,
+                    "name": "#secret",
+                    "flood_policy": "deny",
+                    "transport_key": _b64key("#secret"),
+                    "parent_id": None,
+                }
+            ]
+        ),
+    )
+    # The entry exists and carries the deny flag ...
+    assert [r.flags for r in rm.regions] == [REGION_DENY_FLOOD]
+    # ... so a flood scoped to it still replies plain (find_match returns None).
+    assert rm.find_match(_scoped_flood(get_auto_key_for("#secret")), mask=REGION_DENY_FLOOD) is None
+
+
+def test_wildcard_and_empty_names_are_not_entries():
+    rm = build_region_map(
+        {},
+        _FakeHandler(
+            [
+                {
+                    "id": 3,
+                    "name": "*",
+                    "flood_policy": "allow",
+                    "transport_key": None,
+                    "parent_id": None,
+                },
+                {
+                    "id": 4,
+                    "name": "",
+                    "flood_policy": "allow",
+                    "transport_key": None,
+                    "parent_id": None,
+                },
+                {
+                    "id": 5,
+                    "name": "  ",
+                    "flood_policy": "allow",
+                    "transport_key": None,
+                    "parent_id": None,
+                },
+            ]
+        ),
+    )
+    assert rm.regions == []
+
+
+def test_plain_flood_never_matches():
+    rm = build_region_map(
+        {},
+        _FakeHandler(
+            [
+                {
+                    "id": 1,
+                    "name": "#usa",
+                    "flood_policy": "allow",
+                    "transport_key": _b64key("#usa"),
+                    "parent_id": None,
+                }
+            ]
+        ),
+    )
+    assert rm.find_match(_plain_flood(), mask=REGION_DENY_FLOOD) is None
+
+
+def test_private_region_uses_stored_key():
+    custom = b"\x11" * 16
+    rm = build_region_map(
+        {},
+        _FakeHandler(
+            [
+                {
+                    "id": 7,
+                    "name": "$vip",
+                    "flood_policy": "allow",
+                    "transport_key": base64.b64encode(custom).decode("ascii"),
+                    "parent_id": None,
+                }
+            ]
+        ),
+    )
+    entry = rm.regions[0]
+    assert entry.private_keys == [custom]
+    assert rm.find_match(_scoped_flood(custom), mask=REGION_DENY_FLOOD) is not None
+
+
+def test_private_region_without_key_matches_nothing():
+    rm = build_region_map(
+        {},
+        _FakeHandler(
+            [
+                {
+                    "id": 8,
+                    "name": "$vip",
+                    "flood_policy": "allow",
+                    "transport_key": None,
+                    "parent_id": None,
+                }
+            ]
+        ),
+    )
+    # No usable key for a "$" region -> never matches (core never name-hashes it).
+    assert rm.find_match(_scoped_flood(get_auto_key_for("$vip")), mask=REGION_DENY_FLOOD) is None
+
+
+def test_public_region_with_custom_key_carries_private_key():
+    custom = b"\x22" * 16
+    rm = build_region_map(
+        {},
+        _FakeHandler(
+            [
+                {
+                    "id": 9,
+                    "name": "#usa",
+                    "flood_policy": "allow",
+                    "transport_key": base64.b64encode(custom).decode("ascii"),
+                    "parent_id": None,
+                }
+            ]
+        ),
+    )
+    entry = rm.regions[0]
+    assert entry.private_keys == [custom]  # differs from name hash -> carried through
+    assert rm.find_match(_scoped_flood(custom), mask=REGION_DENY_FLOOD) is not None
+
+
+def test_public_region_with_auto_key_relies_on_name_hash():
+    rm = build_region_map(
+        {},
+        _FakeHandler(
+            [
+                {
+                    "id": 1,
+                    "name": "#usa",
+                    "flood_policy": "allow",
+                    "transport_key": _b64key("#usa"),
+                    "parent_id": None,
+                }
+            ]
+        ),
+    )
+    # Stored key equals the name hash -> not carried as an explicit private key.
+    assert rm.regions[0].private_keys is None
+
+
+def test_missing_storage_yields_empty_map():
+    assert build_region_map({}, None).regions == []
+
+
+# ---------------------------------------------------------------------------
+# Storage change hook fires on transport_keys writes
+# ---------------------------------------------------------------------------
+def test_transport_keys_change_hook_fires_on_writes(tmp_path):
+    handler = SQLiteHandler(tmp_path)
+    cb = MagicMock()
+    handler.set_transport_keys_changed_callback(cb)
+
+    key_id = handler.create_transport_key("#usa", "allow")
+    assert key_id is not None
+    assert cb.call_count == 1
+
+    assert handler.update_transport_key(key_id, flood_policy="deny")
+    assert cb.call_count == 2
+
+    # A no-op update (unknown id) must not fire the hook.
+    assert not handler.update_transport_key(999999, flood_policy="allow")
+    assert cb.call_count == 2
+
+    assert handler.delete_transport_key(key_id)
+    assert cb.call_count == 3
+
+    handler.sync_transport_keys(
+        [
+            {"node_id": "n1", "name": "#eu", "flood_policy": "allow"},
+        ]
+    )
+    assert cb.call_count == 4
+
+
+# ---------------------------------------------------------------------------
+# Daemon wiring: dispatcher + bridges get a shared, non-None map that a runtime
+# region change refreshes for every holder.
+# ---------------------------------------------------------------------------
+def test_daemon_wires_and_refreshes_region_map(tmp_path):
+    handler = SQLiteHandler(tmp_path)
+    handler.create_transport_key("#usa", "allow")
+
+    daemon = RepeaterDaemon({"logging": {}, "mesh": {}})
+    daemon.repeater_handler = SimpleNamespace(storage=SimpleNamespace(sqlite_handler=handler))
+    daemon.dispatcher = SimpleNamespace(region_map=None)
+    daemon.companion_bridges = {1: SimpleNamespace(region_map=None)}
+
+    daemon._init_region_map()
+
+    # Dispatcher has a non-None map with the served region.
+    assert daemon.dispatcher.region_map is not None
+    assert [r.name for r in daemon.dispatcher.region_map.regions] == ["#usa"]
+
+    # A runtime add fires the storage hook -> refresh reaches dispatcher + bridges.
+    handler.create_transport_key("#eu", "allow")
+    assert sorted(r.name for r in daemon.dispatcher.region_map.regions) == ["#eu", "#usa"]
+    assert daemon.companion_bridges[1].region_map is daemon.dispatcher.region_map
+
+    # A runtime delete also reaches every holder.
+    eu = next(r for r in handler.get_transport_keys() if r["name"] == "#eu")
+    handler.delete_transport_key(eu["id"])
+    assert [r.name for r in daemon.dispatcher.region_map.regions] == ["#usa"]
+    assert daemon.companion_bridges[1].region_map is daemon.dispatcher.region_map