feat: implement neighbor discovery session management and API endpoints

repeater/handler_helpers/discovery.py

@@ -8,6 +8,10 @@ allowing other nodes to discover repeaters on the mesh network.
 import asyncio
 import logging
 import secrets
+import threading
+import time
+import uuid
+from typing import Any, Callable, Optional
 
 from openhop_core.node.handlers.control import ControlHandler
 
@@ -22,6 +26,15 @@ logger = logging.getLogger("DiscoveryHelper")
 # 60s (firmware pending_discover_until = futureMillis(60000)).
 DEFAULT_DISCOVERY_RESPONSE_JITTER_MS = 2000
 
+DEFAULT_DISCOVERY_TIMEOUT_SECONDS = 10.0
+DISCOVERY_EVENT_BACKLOG_LIMIT = 512
+
+NODE_TYPE_NAMES = {
+    1: "Chat Node",
+    2: "Repeater",
+    3: "Room Server",
+}
+
 
 class DiscoveryHelper:
     """Helper class for processing discovery requests in the repeater."""
@@ -60,11 +73,257 @@ class DiscoveryHelper:
             debug_log_fn=debug_log_fn,
         )
         self._pending_tasks = set()
+        self._sessions: dict[str, dict[str, Any]] = {}
+        self._sessions_lock = threading.Lock()
 
         # Set up the request callback
         self.control_handler.set_request_callback(self._on_discovery_request)
         logger.debug("Discovery handler initialized")
 
+    def create_session(
+        self,
+        *,
+        timeout: float = DEFAULT_DISCOVERY_TIMEOUT_SECONDS,
+        filter_mask: int,
+        since: int = 0,
+        prefix_only: bool = False,
+        result_enricher: Optional[Callable[[dict[str, Any]], dict[str, Any]]] = None,
+    ) -> dict[str, Any]:
+        """Create a new discovery session and return its public metadata."""
+        session_id = uuid.uuid4().hex
+        tag = secrets.randbits(32)
+        created_at = time.time()
+        session = {
+            "session_id": session_id,
+            "tag": tag,
+            "timeout": max(1.0, float(timeout)),
+            "filter_mask": int(filter_mask) & 0xFF,
+            "since": max(0, int(since)),
+            "prefix_only": bool(prefix_only),
+            "created_at": created_at,
+            "started_at": None,
+            "completed_at": None,
+            "status": "created",
+            "results": {},
+            "events": [],
+            "next_event_id": 1,
+            "error": None,
+            "result_enricher": result_enricher,
+        }
+        with self._sessions_lock:
+            self._sessions[session_id] = session
+        return self.get_session_snapshot(session_id) or {}
+
+    def get_session_snapshot(self, session_id: str) -> Optional[dict[str, Any]]:
+        """Return a public snapshot for a discovery session."""
+        with self._sessions_lock:
+            session = self._sessions.get(session_id)
+            if not session:
+                return None
+            return self._public_session_snapshot(session)
+
+    def get_events_since(self, session_id: str, last_event_id: int = 0) -> Optional[dict[str, Any]]:
+        """Return all session events newer than last_event_id."""
+        with self._sessions_lock:
+            session = self._sessions.get(session_id)
+            if not session:
+                return None
+            events = [event for event in session["events"] if event["id"] > last_event_id]
+            return {
+                "events": events,
+                "status": session["status"],
+                "completed": session["status"] in {"completed", "timed_out", "error", "cancelled"},
+                "latest_event_id": session["next_event_id"] - 1,
+            }
+
+    async def execute_session(self, session_id: str) -> None:
+        """Send a discovery request and stream responses into the session."""
+        session = self._get_session(session_id)
+        if not session:
+            raise ValueError(f"Unknown discovery session: {session_id}")
+
+        if session["status"] != "created":
+            return
+
+        session["started_at"] = time.time()
+        session["status"] = "running"
+        self._emit_event(
+            session_id,
+            "started",
+            {
+                "session_id": session_id,
+                "tag": session["tag"],
+                "timeout": session["timeout"],
+                "filter_mask": session["filter_mask"],
+                "since": session["since"],
+                "prefix_only": session["prefix_only"],
+                "started_at": session["started_at"],
+            },
+        )
+
+        try:
+            from openhop_core.protocol.packet_builder import PacketBuilder
+
+            packet = PacketBuilder.create_discovery_request(
+                tag=session["tag"],
+                filter_mask=session["filter_mask"],
+                since=session["since"],
+                prefix_only=session["prefix_only"],
+            )
+
+            def _response_callback(response_data: dict[str, Any]) -> None:
+                self._record_response(session_id, response_data)
+
+            self.control_handler.set_response_callback(session["tag"], _response_callback)
+
+            if not self.packet_injector:
+                raise RuntimeError("No packet injector available")
+
+            success = await self.packet_injector(packet, wait_for_ack=False)
+            if not success:
+                raise RuntimeError("Failed to send discovery request")
+
+            logger.info(
+                "Discovery request sent for session %s tag 0x%08X filter=0x%02X",
+                session_id,
+                session["tag"],
+                session["filter_mask"],
+            )
+
+            await asyncio.sleep(session["timeout"])
+            self._finish_session(session_id, "completed")
+        except asyncio.CancelledError:
+            self._finish_session(session_id, "cancelled")
+            raise
+        except Exception as e:
+            logger.error("Discovery session %s failed: %s", session_id, e, exc_info=True)
+            self._finish_session(session_id, "error", error=str(e))
+        finally:
+            self.control_handler.clear_response_callback(session["tag"])
+
+    def start_session_task(self, session_id: str) -> None:
+        """Schedule a discovery session on the current event loop."""
+        task = asyncio.create_task(self.execute_session(session_id))
+        self._track_task(task)
+
+    def cleanup_sessions(self, max_age_seconds: int = 120) -> None:
+        """Remove old completed sessions to keep memory bounded."""
+        cutoff = time.time() - max_age_seconds
+        with self._sessions_lock:
+            stale_ids = [
+                session_id
+                for session_id, session in self._sessions.items()
+                if session["status"] in {"completed", "timed_out", "error", "cancelled"}
+                and (session.get("completed_at") or session.get("created_at", 0)) < cutoff
+            ]
+            for session_id in stale_ids:
+                self._sessions.pop(session_id, None)
+
+    def _get_session(self, session_id: str) -> Optional[dict[str, Any]]:
+        with self._sessions_lock:
+            return self._sessions.get(session_id)
+
+    def _record_response(self, session_id: str, response_data: dict[str, Any]) -> None:
+        with self._sessions_lock:
+            session = self._sessions.get(session_id)
+            if not session or session["status"] != "running":
+                return
+
+            result = dict(response_data)
+            result["node_type_name"] = NODE_TYPE_NAMES.get(
+                result.get("node_type"), f"Unknown({result.get('node_type', 0)})"
+            )
+            result["discovered_at"] = time.time()
+
+            enricher = session.get("result_enricher")
+            if enricher:
+                try:
+                    result = enricher(result)
+                except Exception as e:
+                    logger.debug("Discovery result enrichment failed: %s", e)
+
+            result_key = str(result.get("pub_key") or "")
+            if not result_key:
+                return
+
+            existing = session["results"].get(result_key)
+            session["results"][result_key] = result
+            payload = {
+                "session_id": session_id,
+                "tag": session["tag"],
+                "result": result,
+                "is_update": existing is not None,
+                "count": len(session["results"]),
+            }
+            self._append_event_unlocked(session, "discovery_result", payload)
+
+    def _finish_session(self, session_id: str, status: str, error: Optional[str] = None) -> None:
+        with self._sessions_lock:
+            session = self._sessions.get(session_id)
+            if not session or session["status"] in {"completed", "timed_out", "error", "cancelled"}:
+                return
+
+            session["completed_at"] = time.time()
+            session["status"] = status
+            session["error"] = error
+            payload = {
+                "session_id": session_id,
+                "tag": session["tag"],
+                "status": status,
+                "error": error,
+                "count": len(session["results"]),
+                "duration_ms": round(
+                    (
+                        (session["completed_at"] or session["created_at"])
+                        - (session["started_at"] or session["created_at"])
+                    )
+                    * 1000,
+                    2,
+                ),
+                "completed_at": session["completed_at"],
+                "results": list(session["results"].values()),
+            }
+            event_type = "error" if status == "error" else "completed"
+            self._append_event_unlocked(session, event_type, payload)
+
+    def _emit_event(self, session_id: str, event_type: str, payload: dict[str, Any]) -> None:
+        with self._sessions_lock:
+            session = self._sessions.get(session_id)
+            if not session:
+                return
+            self._append_event_unlocked(session, event_type, payload)
+
+    def _append_event_unlocked(
+        self, session: dict[str, Any], event_type: str, payload: dict[str, Any]
+    ) -> None:
+        event_id = session["next_event_id"]
+        session["next_event_id"] += 1
+        session["events"].append(
+            {
+                "id": event_id,
+                "event": event_type,
+                "data": payload,
+            }
+        )
+        if len(session["events"]) > DISCOVERY_EVENT_BACKLOG_LIMIT:
+            session["events"] = session["events"][-DISCOVERY_EVENT_BACKLOG_LIMIT:]
+
+    def _public_session_snapshot(self, session: dict[str, Any]) -> dict[str, Any]:
+        return {
+            "session_id": session["session_id"],
+            "tag": session["tag"],
+            "status": session["status"],
+            "timeout": session["timeout"],
+            "filter_mask": session["filter_mask"],
+            "since": session["since"],
+            "prefix_only": session["prefix_only"],
+            "created_at": session["created_at"],
+            "started_at": session["started_at"],
+            "completed_at": session["completed_at"],
+            "count": len(session["results"]),
+            "error": session["error"],
+        }
+
     def _track_task(self, task: asyncio.Task) -> None:
         self._pending_tasks.add(task)
 

repeater/web/api_endpoints.py

@@ -122,6 +122,9 @@ POLICY_GROUP_KINDS = {
 # GET    /api/unscoped_flood_policy - Get unscoped flood policy
 # POST   /api/unscoped_flood_policy - Update unscoped flood policy
 # POST   /api/ping_neighbor - Ping a neighbor node
+# POST   /api/discover_neighbors_start - Start a live repeater discovery session
+# GET    /api/discover_neighbors_stream?session_id=X - Stream repeater discovery results over SSE
+# POST   /api/add_discovered_neighbor - Persist a discovered node into the neighbors/adverts table
 
 # Identity Management
 # GET    /api/identities - List all identities
@@ -303,6 +306,81 @@ class APIEndpoints:
         end_time = int(time.time())
         return end_time - (hours * 3600), end_time
 
+    def _encode_sse_event(
+        self, payload, event_name: Optional[str] = None, event_id: Optional[int] = None
+    ) -> str:
+        lines = []
+        if event_name:
+            lines.append(f"event: {event_name}")
+        if event_id is not None:
+            lines.append(f"id: {event_id}")
+        lines.append(f"data: {json.dumps(payload, default=str)}")
+        return "\n".join(lines) + "\n\n"
+
+    @staticmethod
+    def _normalize_discovery_node_name(value) -> Optional[str]:
+        """Normalize placeholder discovery names to None.
+
+        Some peers explicitly advertise "Unknown"-style placeholder names.
+        Treat those as missing so callers can fall back to hash/prefix labels.
+        """
+        text = str(value or "").strip()
+        if not text:
+            return None
+
+        normalized = text.lower()
+        if normalized in {"unknown", "unknown node"}:
+            return None
+
+        return text
+
+    def _enrich_discovery_result(self, result: dict) -> dict:
+        enriched = dict(result)
+        enriched["node_name"] = self._normalize_discovery_node_name(enriched.get("node_name"))
+        pub_key = str(enriched.get("pub_key") or "").lower()
+        if not pub_key:
+            return enriched
+
+        try:
+            enriched["node_hash"] = self._fmt_hash(bytes.fromhex(pub_key))
+        except ValueError:
+            enriched["node_hash"] = None
+
+        try:
+            storage = self._get_storage()
+        except Exception:
+            storage = None
+
+        if not storage:
+            enriched["known_neighbor"] = False
+            return enriched
+
+        neighbor_info = {}
+        try:
+            if hasattr(storage, "get_neighbors"):
+                neighbor_info = storage.get_neighbors().get(pub_key, {}) or {}
+        except Exception as exc:
+            logger.debug("Discovery enrichment could not load neighbors: %s", exc)
+
+        if not neighbor_info:
+            try:
+                node_name = storage.get_node_name_by_pubkey(pub_key)
+            except Exception:
+                node_name = None
+            node_name = self._normalize_discovery_node_name(node_name)
+            enriched["known_neighbor"] = bool(node_name)
+            if node_name:
+                enriched["node_name"] = node_name
+            return enriched
+
+        enriched["known_neighbor"] = True
+        enriched["node_name"] = self._normalize_discovery_node_name(neighbor_info.get("node_name"))
+        enriched["contact_type"] = neighbor_info.get("contact_type")
+        enriched["zero_hop"] = neighbor_info.get("zero_hop")
+        enriched["last_seen"] = neighbor_info.get("last_seen")
+        enriched["advert_count"] = neighbor_info.get("advert_count")
+        return enriched
+
     def _process_counter_data(self, data_points, timestamps_ms):
         rates = []
         prev_value = None
@@ -4044,6 +4122,222 @@ class APIEndpoints:
             logger.error(f"Error pinging neighbor: {e}", exc_info=True)
             return self._error(str(e))
 
+    @cherrypy.expose
+    @cherrypy.tools.json_out()
+    @cherrypy.tools.json_in()
+    def discover_neighbors_start(self):
+
+        self._set_cors_headers()
+
+        if cherrypy.request.method == "OPTIONS":
+            return ""
+
+        try:
+            self._require_post()
+            data = cherrypy.request.json or {}
+            timeout = float(data.get("timeout", 5))
+            filter_mask = int(data.get("filter_mask", 1 << 2))
+            since = int(data.get("since", 0))
+            prefix_only = bool(data.get("prefix_only", False))
+
+            if timeout < 1 or timeout > 60:
+                return self._error("timeout must be between 1 and 60 seconds")
+            if filter_mask < 0 or filter_mask > 0xFF:
+                return self._error("filter_mask must be between 0x00 and 0xFF")
+            if since < 0:
+                return self._error("since must be non-negative")
+
+            if self.event_loop is None:
+                return self._error("Event loop not available")
+
+            discovery_helper = getattr(self.daemon_instance, "discovery_helper", None)
+            if not discovery_helper:
+                return self._error("Discovery helper not available")
+
+            discovery_helper.cleanup_sessions()
+            session = discovery_helper.create_session(
+                timeout=timeout,
+                filter_mask=filter_mask,
+                since=since,
+                prefix_only=prefix_only,
+                result_enricher=self._enrich_discovery_result,
+            )
+
+            self.event_loop.call_soon_threadsafe(
+                discovery_helper.start_session_task, session["session_id"]
+            )
+
+            return self._success(
+                session,
+                message="Discovery session started",
+            )
+        except cherrypy.HTTPError:
+            raise
+        except Exception as e:
+            logger.error("Error starting discovery session: %s", e, exc_info=True)
+            return self._error(str(e))
+
+    @cherrypy.expose
+    def discover_neighbors_stream(self, session_id=None, last_event_id: Optional[str] = None):
+        self._set_cors_headers()
+        cherrypy.response.headers["Content-Type"] = "text/event-stream"
+        cherrypy.response.headers["Cache-Control"] = "no-cache"
+        cherrypy.response.headers["Connection"] = "keep-alive"
+        cherrypy.response.headers["X-Accel-Buffering"] = "no"
+
+        discovery_helper = getattr(self.daemon_instance, "discovery_helper", None)
+
+        try:
+            cursor = int(last_event_id) if last_event_id is not None else 0
+        except (TypeError, ValueError):
+            cursor = 0
+
+        def generate():
+            if not session_id:
+                yield self._encode_sse_event(
+                    {"type": "error", "error": "Missing session_id"},
+                    event_name="error",
+                )
+                return
+
+            if not discovery_helper:
+                yield self._encode_sse_event(
+                    {"type": "error", "error": "Discovery helper not available"},
+                    event_name="error",
+                )
+                return
+
+            snapshot = discovery_helper.get_session_snapshot(session_id)
+            if not snapshot:
+                yield self._encode_sse_event(
+                    {"type": "error", "error": f"Unknown discovery session: {session_id}"},
+                    event_name="error",
+                )
+                return
+
+            yield self._encode_sse_event(
+                {
+                    "type": "connected",
+                    "session": snapshot,
+                },
+                event_name="connected",
+            )
+
+            current_cursor = cursor
+            try:
+                while True:
+                    event_state = discovery_helper.get_events_since(session_id, current_cursor)
+                    if event_state is None:
+                        yield self._encode_sse_event(
+                            {
+                                "type": "error",
+                                "error": f"Unknown discovery session: {session_id}",
+                            },
+                            event_name="error",
+                        )
+                        return
+
+                    events = event_state.get("events", [])
+                    if events:
+                        for event in events:
+                            current_cursor = max(current_cursor, int(event.get("id", 0)))
+                            yield self._encode_sse_event(
+                                event.get("data", {}),
+                                event_name=event.get("event"),
+                                event_id=event.get("id"),
+                            )
+                        if event_state.get("completed"):
+                            return
+                    else:
+                        if event_state.get("completed"):
+                            return
+                        yield self._encode_sse_event(
+                            {"type": "keepalive", "session_id": session_id},
+                            event_name="keepalive",
+                            event_id=current_cursor if current_cursor > 0 else None,
+                        )
+
+                    time.sleep(0.5)
+            except GeneratorExit:
+                logger.debug("Discovery SSE stream closed by client for session %s", session_id)
+            except Exception as exc:
+                logger.error("Discovery SSE stream error: %s", exc, exc_info=True)
+                yield self._encode_sse_event(
+                    {"type": "error", "error": str(exc), "session_id": session_id},
+                    event_name="error",
+                    event_id=current_cursor if current_cursor > 0 else None,
+                )
+
+        return generate()
+
+    discover_neighbors_stream._cp_config = {"response.stream": True}
+
+    @cherrypy.expose
+    @cherrypy.tools.json_out()
+    @cherrypy.tools.json_in()
+    @require_auth
+    def add_discovered_neighbor(self):
+
+        self._set_cors_headers()
+
+        if cherrypy.request.method == "OPTIONS":
+            return ""
+
+        try:
+            self._require_post()
+            data = cherrypy.request.json or {}
+            pub_key = str(data.get("pub_key") or "").strip().lower()
+            node_name = self._normalize_discovery_node_name(data.get("node_name"))
+            node_type = int(data.get("node_type", 0))
+            rssi = data.get("rssi")
+            snr = data.get("response_snr", data.get("snr"))
+
+            if not pub_key:
+                return self._error("pub_key is required")
+            if not re.fullmatch(r"[0-9a-f]{16}|[0-9a-f]{64}", pub_key):
+                return self._error("pub_key must be 8-byte or 32-byte hex")
+
+            contact_type = {
+                1: "Chat Node",
+                2: "Repeater",
+                3: "Room Server",
+            }.get(node_type, "Unknown")
+
+            advert_record = {
+                "timestamp": time.time(),
+                "pubkey": pub_key,
+                "node_name": node_name,
+                "is_repeater": node_type == 2,
+                "route_type": 2,
+                "contact_type": contact_type,
+                "latitude": None,
+                "longitude": None,
+                "rssi": int(rssi) if rssi is not None else None,
+                "snr": float(snr) if snr is not None else None,
+                "is_new_neighbor": True,
+                "zero_hop": True,
+            }
+
+            storage = self._get_storage()
+            storage.record_advert(advert_record)
+
+            enriched = self._enrich_discovery_result(
+                {
+                    "pub_key": pub_key,
+                    "node_name": node_name,
+                    "node_type": node_type,
+                    "node_type_name": contact_type,
+                    "rssi": advert_record["rssi"],
+                    "response_snr": advert_record["snr"],
+                }
+            )
+            return self._success(enriched, message="Neighbor added to adverts database")
+        except cherrypy.HTTPError:
+            raise
+        except Exception as e:
+            logger.error("Error adding discovered neighbor: %s", e, exc_info=True)
+            return self._error(str(e))
+
     # ========== Identity Management Endpoints ==========
 
     @cherrypy.expose

repeater/web/auth/middleware.py

@@ -39,6 +39,26 @@ def require_auth(func):
             else:
                 logger.warning("Invalid or expired JWT token")
 
+        request_params = getattr(cherrypy.request, "params", None)
+        if request_params is None:
+            request_params = {}
+
+        query_token = request_params.get("token")
+        if query_token:
+            payload = jwt_handler.verify_jwt(query_token)
+
+            if payload:
+                cherrypy.request.user = {
+                    "username": payload["sub"],
+                    "client_id": payload["client_id"],
+                    "auth_type": "jwt_query",
+                }
+                if hasattr(cherrypy.request, "params") and "token" in cherrypy.request.params:
+                    del cherrypy.request.params["token"]
+                return func(*args, **kwargs)
+            else:
+                logger.warning("Invalid or expired JWT query token")
+
         # Try API token authentication
         api_key = cherrypy.request.headers.get("X-API-Key", "")
         if api_key:

repeater/web/openapi.yaml

@@ -1600,6 +1600,163 @@ paths:
                   error:
                     type: string
 
+  /discover_neighbors_start:
+    post:
+      tags: [Network Policy]
+      summary: Start neighbor discovery session
+      description: Start a short-lived discovery broadcast session and return a session ID for SSE streaming.
+      requestBody:
+        required: false
+        content:
+          application/json:
+            schema:
+              type: object
+              properties:
+                timeout:
+                  type: number
+                  minimum: 1
+                  maximum: 60
+                  default: 5
+                filter_mask:
+                  type: integer
+                  minimum: 0
+                  maximum: 255
+                  default: 4
+                since:
+                  type: integer
+                  minimum: 0
+                  default: 0
+                prefix_only:
+                  type: boolean
+                  default: false
+      responses:
+        '200':
+          description: Discovery session started
+          content:
+            application/json:
+              schema:
+                type: object
+                properties:
+                  success:
+                    type: boolean
+                  data:
+                    type: object
+                    properties:
+                      session_id:
+                        type: string
+                      tag:
+                        type: integer
+                      status:
+                        type: string
+                      timeout:
+                        type: number
+                      filter_mask:
+                        type: integer
+                      since:
+                        type: integer
+                      prefix_only:
+                        type: boolean
+                      created_at:
+                        type: number
+                      started_at:
+                        type: number
+                        nullable: true
+                      completed_at:
+                        type: number
+                        nullable: true
+                      count:
+                        type: integer
+                      error:
+                        type: string
+                        nullable: true
+                  message:
+                    type: string
+                  error:
+                    type: string
+
+  /discover_neighbors_stream:
+    get:
+      tags: [Network Policy]
+      summary: Stream discovery events
+      description: Stream live neighbor discovery session events via Server-Sent Events (SSE).
+      security:
+        - BearerAuth: []
+        - ApiKeyAuth: []
+      parameters:
+        - name: session_id
+          in: query
+          required: true
+          schema:
+            type: string
+        - name: last_event_id
+          in: query
+          required: false
+          schema:
+            type: integer
+        - name: token
+          in: query
+          required: false
+          schema:
+            type: string
+          description: JWT token for EventSource clients that cannot send Authorization headers.
+      responses:
+        '200':
+          description: SSE event stream
+          content:
+            text/event-stream:
+              schema:
+                type: string
+
+  /add_discovered_neighbor:
+    post:
+      tags: [Network Policy]
+      summary: Add discovered node as neighbor
+      description: Persist a discovered node into the adverts/neighbors store for future management.
+      security:
+        - BearerAuth: []
+        - ApiKeyAuth: []
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              type: object
+              required: [pub_key]
+              properties:
+                pub_key:
+                  type: string
+                  description: Hex pubkey (8-byte or 32-byte form).
+                node_name:
+                  type: string
+                  nullable: true
+                node_type:
+                  type: integer
+                rssi:
+                  type: integer
+                  nullable: true
+                response_snr:
+                  type: number
+                  nullable: true
+                snr:
+                  type: number
+                  nullable: true
+      responses:
+        '200':
+          description: Node added
+          content:
+            application/json:
+              schema:
+                type: object
+                properties:
+                  success:
+                    type: boolean
+                  data:
+                    type: object
+                  message:
+                    type: string
+                  error:
+                    type: string
+
   /policy:
     get:
       tags: [Network Policy]

tests/test_api_endpoints_core_coverage.py

@@ -25,6 +25,61 @@ def _attach_storage(api, storage):
     api.daemon_instance = SimpleNamespace(repeater_handler=SimpleNamespace(storage=storage))
 
 
+class _FakeDiscoveryHelper:
+    def __init__(self):
+        self.cleanup_called = False
+        self.started_sessions = []
+        self._session = {
+            "session_id": "sess-1",
+            "tag": 123,
+            "status": "created",
+            "timeout": 5.0,
+            "filter_mask": 0x04,
+            "since": 0,
+            "prefix_only": False,
+            "created_at": 1.0,
+            "started_at": None,
+            "completed_at": None,
+            "count": 0,
+            "error": None,
+        }
+        self._events = [
+            {"id": 1, "event": "started", "data": {"session_id": "sess-1"}},
+            {
+                "id": 2,
+                "event": "discovery_result",
+                "data": {"session_id": "sess-1", "result": {"pub_key": "aa" * 32}},
+            },
+            {"id": 3, "event": "completed", "data": {"session_id": "sess-1", "count": 1}},
+        ]
+
+    def cleanup_sessions(self):
+        self.cleanup_called = True
+
+    def create_session(self, **kwargs):
+        self._session.update(
+            {
+                "timeout": kwargs["timeout"],
+                "filter_mask": kwargs["filter_mask"],
+                "since": kwargs["since"],
+                "prefix_only": kwargs["prefix_only"],
+            }
+        )
+        return dict(self._session)
+
+    def start_session_task(self, session_id):
+        self.started_sessions.append(session_id)
+
+    def get_session_snapshot(self, session_id):
+        return dict(self._session) if session_id == self._session["session_id"] else None
+
+    def get_events_since(self, session_id, last_event_id=0):
+        if session_id != self._session["session_id"]:
+            return None
+        events = [event for event in self._events if event["id"] > last_event_id]
+        return {"events": events, "completed": True, "status": "completed", "latest_event_id": 3}
+
+
 @pytest.fixture
 def cherrypy_ctx(monkeypatch):
     request = SimpleNamespace(method="GET", params={}, json={})
@@ -148,6 +203,134 @@ def test_success_and_error_helpers():
     assert err == {"success": False, "error": "boom"}
 
 
+def test_discover_neighbors_start_schedules_session(cherrypy_ctx):
+    request, _ = cherrypy_ctx
+    request.method = "POST"
+    request.json = {"timeout": 7, "filter_mask": 0x04, "since": 0, "prefix_only": False}
+
+    helper = _FakeDiscoveryHelper()
+    loop = MagicMock()
+    api = _make_api()
+    api.event_loop = loop
+    api.daemon_instance = SimpleNamespace(discovery_helper=helper)
+
+    result = api.discover_neighbors_start()
+
+    assert result["success"] is True
+    assert result["data"]["session_id"] == "sess-1"
+    assert helper.cleanup_called is True
+    loop.call_soon_threadsafe.assert_called_once()
+
+
+def test_discover_neighbors_stream_yields_session_events(cherrypy_ctx):
+    request, response = cherrypy_ctx
+    request.method = "GET"
+
+    helper = _FakeDiscoveryHelper()
+    api = _make_api()
+    api.daemon_instance = SimpleNamespace(discovery_helper=helper)
+
+    stream = api.discover_neighbors_stream(session_id="sess-1")
+    payload = "".join(list(stream))
+
+    assert response.headers["Content-Type"] == "text/event-stream"
+    assert "event: connected" in payload
+    assert "event: started" in payload
+    assert "event: discovery_result" in payload
+    assert "event: completed" in payload
+
+
+def test_add_discovered_neighbor_records_zero_hop_advert(cherrypy_ctx, monkeypatch):
+    request, _ = cherrypy_ctx
+    request.method = "POST"
+    request.headers = {"Authorization": "Bearer test-token"}
+    request.path_info = "/api/add_discovered_neighbor"
+    request.json = {
+        "pub_key": "aa" * 32,
+        "node_name": "Field Repeater",
+        "node_type": 2,
+        "rssi": -71,
+        "response_snr": 4.5,
+    }
+
+    jwt_handler = MagicMock()
+    jwt_handler.verify_jwt.return_value = {"sub": "tester", "client_id": "ui"}
+    token_manager = MagicMock()
+    monkeypatch.setattr(
+        cherrypy,
+        "config",
+        {"jwt_handler": jwt_handler, "token_manager": token_manager},
+        raising=False,
+    )
+
+    storage = MagicMock()
+    api = _make_api()
+    api._enrich_discovery_result = lambda result: {**result, "known_neighbor": True}
+    _attach_storage(api, storage)
+
+    result = api.add_discovered_neighbor()
+
+    assert result["success"] is True
+    storage.record_advert.assert_called_once()
+    advert_record = storage.record_advert.call_args.args[0]
+    assert advert_record["route_type"] == 2
+    assert advert_record["zero_hop"] is True
+    assert advert_record["contact_type"] == "Repeater"
+
+
+def test_add_discovered_neighbor_normalizes_unknown_name(cherrypy_ctx, monkeypatch):
+    request, _ = cherrypy_ctx
+    request.method = "POST"
+    request.headers = {"Authorization": "Bearer test-token"}
+    request.path_info = "/api/add_discovered_neighbor"
+    request.json = {
+        "pub_key": "bb" * 32,
+        "node_name": "Unknown Node",
+        "node_type": 2,
+    }
+
+    jwt_handler = MagicMock()
+    jwt_handler.verify_jwt.return_value = {"sub": "tester", "client_id": "ui"}
+    token_manager = MagicMock()
+    monkeypatch.setattr(
+        cherrypy,
+        "config",
+        {"jwt_handler": jwt_handler, "token_manager": token_manager},
+        raising=False,
+    )
+
+    storage = MagicMock()
+    api = _make_api()
+    api._enrich_discovery_result = lambda result: result
+    _attach_storage(api, storage)
+
+    result = api.add_discovered_neighbor()
+
+    assert result["success"] is True
+    advert_record = storage.record_advert.call_args.args[0]
+    assert advert_record["node_name"] is None
+
+
+def test_enrich_discovery_result_treats_placeholder_name_as_unknown():
+    api = _make_api()
+
+    storage = MagicMock()
+    storage.get_neighbors.return_value = {}
+    storage.get_node_name_by_pubkey.return_value = "Unknown"
+    _attach_storage(api, storage)
+
+    result = api._enrich_discovery_result(
+        {
+            "pub_key": "cc" * 32,
+            "node_name": "Unknown Node",
+            "node_type": 2,
+        }
+    )
+
+    assert result["known_neighbor"] is False
+    assert result["node_name"] is None
+
+
 def test_get_time_range_uses_current_time(monkeypatch):
     api = _make_api()
     monkeypatch.setattr("repeater.web.api_endpoints.time.time", lambda: 10_000)

tests/test_handler_helpers_trace_discovery_login.py

@@ -254,6 +254,120 @@ async def test_discovery_response_jitter_disabled_does_not_sleep():
     injector.assert_awaited_once()
 
 
+@pytest.mark.asyncio
+async def test_discovery_session_collects_and_completes():
+    injector = AsyncMock(return_value=True)
+    helper = DiscoveryHelper(
+        local_identity=FakeIdentity(0x42),
+        packet_injector=injector,
+        response_jitter_ms=0,
+    )
+
+    session = helper.create_session(timeout=0.01, filter_mask=0x04)
+    session_id = session["session_id"]
+
+    execute_task = asyncio.create_task(helper.execute_session(session_id))
+    await asyncio.sleep(0)
+
+    tag = helper.get_session_snapshot(session_id)["tag"]
+    callback = helper.control_handler._response_callbacks[tag]
+    callback(
+        {
+            "tag": tag,
+            "node_type": 2,
+            "inbound_snr": 1.0,
+            "response_snr": 2.0,
+            "rssi": -70,
+            "pub_key": "aa" * 32,
+            "timestamp": 123.0,
+            "valid": True,
+        }
+    )
+
+    await execute_task
+
+    snapshot = helper.get_session_snapshot(session_id)
+    assert snapshot["status"] == "completed"
+    assert snapshot["count"] == 1
+
+    event_state = helper.get_events_since(session_id)
+    event_names = [event["event"] for event in event_state["events"]]
+    assert "started" in event_names
+    assert "discovery_result" in event_names
+    assert "completed" in event_names
+
+
+@pytest.mark.asyncio
+async def test_discovery_session_deduplicates_by_pubkey():
+    helper = DiscoveryHelper(
+        local_identity=FakeIdentity(0x42),
+        packet_injector=AsyncMock(return_value=True),
+        response_jitter_ms=0,
+    )
+    session = helper.create_session(timeout=1, filter_mask=0x04)
+    session_id = session["session_id"]
+
+    helper._record_response(
+        session_id,
+        {
+            "tag": session["tag"],
+            "node_type": 2,
+            "inbound_snr": 1.0,
+            "response_snr": 2.0,
+            "rssi": -70,
+            "pub_key": "bb" * 32,
+            "timestamp": 1.0,
+        },
+    )
+    helper._record_response(
+        session_id,
+        {
+            "tag": session["tag"],
+            "node_type": 2,
+            "inbound_snr": 1.5,
+            "response_snr": 2.5,
+            "rssi": -60,
+            "pub_key": "bb" * 32,
+            "timestamp": 2.0,
+        },
+    )
+
+    snapshot = helper.get_session_snapshot(session_id)
+    assert snapshot["count"] == 0  # session not running yet, responses ignored
+
+    helper._get_session(session_id)["status"] = "running"
+    helper._record_response(
+        session_id,
+        {
+            "tag": session["tag"],
+            "node_type": 2,
+            "inbound_snr": 1.0,
+            "response_snr": 2.0,
+            "rssi": -70,
+            "pub_key": "cc" * 32,
+            "timestamp": 1.0,
+        },
+    )
+    helper._record_response(
+        session_id,
+        {
+            "tag": session["tag"],
+            "node_type": 2,
+            "inbound_snr": 1.5,
+            "response_snr": 2.5,
+            "rssi": -60,
+            "pub_key": "cc" * 32,
+            "timestamp": 2.0,
+        },
+    )
+
+    snapshot = helper.get_session_snapshot(session_id)
+    assert snapshot["count"] == 1
+    latest_events = helper.get_events_since(session_id)["events"]
+    result_events = [event for event in latest_events if event["event"] == "discovery_result"]
+    assert result_events[-1]["data"]["is_update"] is True
+
+
 def test_discovery_send_response_without_injector_is_safe():
     helper = DiscoveryHelper(local_identity=FakeIdentity(0x42), packet_injector=None)