refactor(repeater): consolidate identity collision validation

The startup preflight re-implemented the identity collision rules in
main.py, reconstructing IdentityManager state by string-parsing its
'type:name' labels, and every configured identity was parsed and
constructed twice (once for preflight, once for loading), duplicating
config warnings on each start.

Move batch validation into IdentityManager.validate_specs(), which
checks a batch of IdentitySpec entries against registered identities
and against each other without mutating state, and relocate
IdentityConfigurationError next to it. The daemon now parses room
server and companion configs once during preflight and the identity
loaders reuse the cached specs.

repeater/identity_manager.py

@@ -1,9 +1,32 @@
 import logging
-from typing import Any, Dict, Optional, Tuple
+from dataclasses import dataclass
+from typing import Any, Dict, Iterable, Optional, Tuple
 
 logger = logging.getLogger("IdentityManager")
 
 
+class IdentityConfigurationError(RuntimeError):
+    """A configured local identity cannot be represented safely."""
+
+
+@dataclass(frozen=True)
+class IdentitySpec:
+    """A parsed-but-unregistered local identity from configuration."""
+
+    name: str
+    identity: Any  # openhop_core LocalIdentity (or compatible)
+    config: dict
+    identity_type: str  # "repeater" | "room_server" | "companion"
+
+    @property
+    def label(self) -> str:
+        return f"{self.identity_type}:{self.name}"
+
+    @property
+    def hash_byte(self) -> int:
+        return self.identity.get_public_key()[0]
+
+
 class IdentityManager:
     def __init__(self, config: dict):
         self.config = config
@@ -38,6 +61,40 @@ class IdentityManager:
 
         return None
 
+    def validate_specs(self, specs: Iterable[IdentitySpec]) -> None:
+        """Raise ``IdentityConfigurationError`` on any name or hash collision.
+
+        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.
+        """
+        batch_hashes: Dict[int, IdentitySpec] = {}
+        batch_names: Dict[str, IdentitySpec] = {}
+
+        for spec in specs:
+            error = self.registration_error(spec.name, spec.identity)
+            if error:
+                raise IdentityConfigurationError(error)
+            existing = batch_names.get(spec.name)
+            if existing is not None:
+                raise IdentityConfigurationError(
+                    f"Local identity name '{spec.name}' conflicts with existing "
+                    f"identity '{existing.label}'"
+                )
+            existing = batch_hashes.get(spec.hash_byte)
+            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"
+                )
+            batch_names[spec.name] = spec
+            batch_hashes[spec.hash_byte] = spec
+
     def validate_identity(self, name: str, identity) -> bool:
         """Log and report whether an identity can be registered without mutation."""
         error = self.registration_error(name, identity)

repeater/main.py

@@ -31,7 +31,7 @@ from repeater.handler_helpers import (
     TextHelper,
     TraceHelper,
 )
-from repeater.identity_manager import IdentityManager
+from repeater.identity_manager import IdentityConfigurationError, IdentityManager, IdentitySpec
 from repeater.packet_router import PacketRouter
 from repeater.sensors import SensorManager
 from repeater.utils_packet import create_scoped_advert_packet
@@ -44,10 +44,6 @@ logger = logging.getLogger("RepeaterDaemon")
 _COMPANION_LOAD_RETRY_DELAY_SEC = 0.5
 
 
-class IdentityConfigurationError(RuntimeError):
-    """A configured local identity cannot be represented safely."""
-
-
 async def _load_companion_rows_verified(
     loader, counter, kind: str, companion_hash_str: str, name: str, **loader_kwargs
 ):
@@ -109,6 +105,10 @@ class RepeaterDaemon:
         self.router = None
         self.companion_bridges: dict[int, object] = {}
         self.companion_frame_servers: list = []
+        # 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
+        self._companion_specs: list[IdentitySpec] | None = None
         self._shutdown_started = False
         self._main_task = None
         self.radio_status = "unknown"
@@ -124,7 +124,7 @@ class RepeaterDaemon:
         _log_buffer.setLevel(getattr(logging, log_level))
         root_logger.addHandler(_log_buffer)
 
-    def _configured_identity_specs(self, identity_type: str) -> list[tuple]:
+    def _configured_identity_specs(self, identity_type: str) -> list[IdentitySpec]:
         """Build valid configured local identities without registering them.
 
         Invalid optional room-server or companion entries retain the existing
@@ -181,58 +181,35 @@ class RepeaterDaemon:
                 logger.error("Failed to create %s identity '%s': %s", label.lower(), name, error)
                 continue
 
-            specs.append((name, identity, identity_config, identity_type))
+            specs.append(
+                IdentitySpec(
+                    name=name,
+                    identity=identity,
+                    config=identity_config,
+                    identity_type=identity_type,
+                )
+            )
 
         return specs
 
-    def _validate_identity_specs(
-        self, specs: list[tuple], *, include_registered: bool = True
-    ) -> None:
-        """Reject duplicate local names or one-byte public-key prefixes."""
-        hashes: dict[int, str] = {}
-        names: dict[str, str] = {}
-
-        if include_registered and self.identity_manager:
-            for hash_byte, (_, _, _) in getattr(self.identity_manager, "identities", {}).items():
-                registered_name = getattr(self.identity_manager, "registered_hashes", {}).get(
-                    hash_byte, "unknown"
-                )
-                hashes[hash_byte] = registered_name
-                if ":" in registered_name:
-                    names[registered_name.split(":", 1)[1]] = registered_name
-
-            for registered_name, (_, _, identity_type) in getattr(
-                self.identity_manager, "named_identities", {}
-            ).items():
-                names.setdefault(registered_name, identity_type)
-
-        for name, identity, _, identity_type in specs:
-            label = f"{identity_type}:{name}"
-            hash_byte = identity.get_public_key()[0]
-            existing_name = names.get(name)
-            if existing_name:
-                raise IdentityConfigurationError(
-                    f"Local identity name '{name}' conflicts with existing identity "
-                    f"'{existing_name}'"
-                )
-            existing_hash = hashes.get(hash_byte)
-            if existing_hash:
-                raise IdentityConfigurationError(
-                    f"Local identity '{label}' (hash=0x{hash_byte:02X}) conflicts "
-                    f"with '{existing_hash}'; local identities must have unique "
-                    "one-byte public-key prefixes"
-                )
-            names[name] = label
-            hashes[hash_byte] = label
-
     def _preflight_configured_local_identities(self, local_identity) -> None:
-        """Validate every configured local identity before stateful setup begins."""
+        """Validate every configured local identity before stateful setup begins.
+
+        The parsed room-server and companion specs are cached so the identity
+        loaders reuse them instead of re-parsing the config (and re-logging
+        every invalid entry). Collision rules live in
+        ``IdentityManager.validate_specs``; at this point the manager holds no
+        registered identities, so this is a pure batch check.
+        """
+        self._room_server_specs = self._configured_identity_specs("room_server")
+        self._companion_specs = self._configured_identity_specs("companion")
         specs = [
-            ("repeater", local_identity, self.config, "repeater"),
-            *self._configured_identity_specs("room_server"),
-            *self._configured_identity_specs("companion"),
+            IdentitySpec("repeater", local_identity, self.config, "repeater"),
+            *self._room_server_specs,
+            *self._companion_specs,
         ]
-        self._validate_identity_specs(specs, include_registered=False)
+        manager = self.identity_manager or IdentityManager(self.config)
+        manager.validate_specs(specs)
 
     async def initialize(self):
 
@@ -626,16 +603,19 @@ class RepeaterDaemon:
             raise
 
     async def _load_additional_identities(self):
-        room_specs = self._configured_identity_specs("room_server")
-        self._validate_identity_specs(room_specs)
+        room_specs = self._room_server_specs
+        if room_specs is None:
+            room_specs = self._configured_identity_specs("room_server")
+        self.identity_manager.validate_specs(room_specs)
 
-        for name, room_identity, room_config, _ in room_specs:
+        for spec in room_specs:
+            name, room_identity = spec.name, spec.identity
             try:
                 # Register with the manager and all helpers
                 success = self._register_identity_everywhere(
                     name=name,
                     identity=room_identity,
-                    config=room_config,
+                    config=spec.config,
                     identity_type="room_server",
                 )
 
@@ -732,13 +712,15 @@ class RepeaterDaemon:
         """Load companion identities from config and create CompanionBridge + frame server for each."""
         from repeater.companion import CompanionFrameServer, RepeaterCompanionBridge
 
-        companion_specs = self._configured_identity_specs("companion")
+        companion_specs = self._companion_specs
+        if companion_specs is None:
+            companion_specs = self._configured_identity_specs("companion")
         if not companion_specs:
             return
 
         # Validate the complete companion set before any bridge can restore or
         # mutate a hash-keyed SQLite namespace, or any TCP server can bind.
-        self._validate_identity_specs(companion_specs)
+        self.identity_manager.validate_specs(companion_specs)
 
         sqlite_handler = None
         if self.repeater_handler and self.repeater_handler.storage:
@@ -754,7 +736,8 @@ class RepeaterDaemon:
             else self.config.get("radio", {})
         )
 
-        for name, identity, comp_config, _ in companion_specs:
+        for spec in companion_specs:
+            name, identity, comp_config = spec.name, spec.identity, spec.config
             try:
                 settings = comp_config.get("settings") or {}
                 pubkey = identity.get_public_key()

tests/test_companion_state_load.py

@@ -147,6 +147,7 @@ class TestCompanionInitSurfacesLoadFailure:
             },
         }
         daemon = RepeaterDaemon(config, radio=object())
+        daemon.identity_manager = IdentityManager({})
         daemon.router = SimpleNamespace(inject_packet=AsyncMock())
         daemon.repeater_handler = SimpleNamespace(
             storage=SimpleNamespace(sqlite_handler=sqlite), radio_config={}

tests/test_identity_collision_preflight.py

@@ -1,3 +1,4 @@
+import logging
 from types import SimpleNamespace
 from unittest.mock import patch
 
@@ -68,6 +69,28 @@ async def test_companion_set_collision_is_rejected_before_bridge_or_server_creat
     assert daemon.companion_frame_servers == []
 
 
+@pytest.mark.asyncio
+async def test_invalid_config_entry_logs_once_across_preflight_and_load(caplog):
+    """Preflight parses the config once and the loaders reuse the cached
+    specs, so an invalid entry produces exactly one error per startup."""
+    daemon = RepeaterDaemon(
+        _config(companions=({"name": "bad", "identity_key": "not-hex"},)),
+        radio=object(),
+    )
+    daemon.identity_manager = IdentityManager({})
+    local_identity = _SeedFirstByteIdentity(b"\x10" * 32)
+
+    with (
+        patch("openhop_core.LocalIdentity", _SeedFirstByteIdentity),
+        caplog.at_level(logging.ERROR, logger="RepeaterDaemon"),
+    ):
+        daemon._preflight_configured_local_identities(local_identity)
+        await daemon._load_companion_identities()
+
+    invalid_key_logs = [r for r in caplog.records if "invalid hex" in r.getMessage()]
+    assert len(invalid_key_logs) == 1
+
+
 @pytest.mark.asyncio
 async def test_hot_added_companion_collision_is_rejected_before_stateful_setup():
     daemon = RepeaterDaemon(_config(), radio=object())

tests/test_identity_manager.py

@@ -1,4 +1,6 @@
-from repeater.identity_manager import IdentityManager
+import pytest
+
+from repeater.identity_manager import IdentityConfigurationError, IdentityManager, IdentitySpec
 
 
 class _FakeIdentity:
@@ -67,3 +69,60 @@ def test_identity_manager_list_handles_none_identity_fields():
     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():
+    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, {}, "repeater"),
+                IdentitySpec("beta", id_b, {}, "companion"),
+            ]
+        )
+
+
+def test_validate_specs_rejects_intra_batch_duplicate_name():
+    mgr = IdentityManager(config={})
+    id_a = _FakeIdentity(bytes([0x11]) + b"A" * 31)
+    id_b = _FakeIdentity(bytes([0x22]) + b"B" * 31)
+
+    with pytest.raises(IdentityConfigurationError, match="repeater:alpha"):
+        mgr.validate_specs(
+            [
+                IdentitySpec("alpha", id_a, {}, "repeater"),
+                IdentitySpec("alpha", id_b, {}, "companion"),
+            ]
+        )
+
+
+def test_validate_specs_rejects_registered_collisions_without_mutation():
+    mgr = IdentityManager(config={})
+    id_a = _FakeIdentity(bytes([0x11]) + b"A" * 31)
+    id_hash_collision = _FakeIdentity(bytes([0x11]) + b"B" * 31)
+    id_name_collision = _FakeIdentity(bytes([0x22]) + b"C" * 31)
+
+    assert mgr.register_identity("alpha", id_a, {}, "repeater") is True
+
+    with pytest.raises(IdentityConfigurationError, match="conflicts"):
+        mgr.validate_specs([IdentitySpec("beta", id_hash_collision, {}, "companion")])
+    with pytest.raises(IdentityConfigurationError, match="already registered"):
+        mgr.validate_specs([IdentitySpec("alpha", id_name_collision, {}, "companion")])
+
+    # Validation never registers anything.
+    assert mgr.get_identity_by_hash(0x22) is None
+    assert mgr.get_identity_by_name("beta") is None
+
+
+def test_validate_specs_accepts_distinct_batch():
+    mgr = IdentityManager(config={})
+    mgr.validate_specs(
+        [
+            IdentitySpec("alpha", _FakeIdentity(bytes([0x11]) + b"A" * 31), {}, "repeater"),
+            IdentitySpec("beta", _FakeIdentity(bytes([0x22]) + b"B" * 31), {}, "room_server"),
+            IdentitySpec("gamma", _FakeIdentity(bytes([0x33]) + b"C" * 31), {}, "companion"),
+        ]
+    )

tests/test_main_py_more.py

@@ -4,6 +4,7 @@ from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
+from repeater.identity_manager import IdentityManager
 from repeater.main import RepeaterDaemon
 
 
@@ -43,7 +44,7 @@ async def test_load_additional_identities_valid_and_invalid_entries():
     }
 
     daemon = RepeaterDaemon(cfg, radio=object())
-    daemon.identity_manager = SimpleNamespace(list_identities=lambda: [1, 2])
+    daemon.identity_manager = IdentityManager({})
     daemon._register_identity_everywhere = MagicMock(return_value=True)
 
     with patch("openhop_core.LocalIdentity", _FakeLocalIdentity):