Step 2

app/decoder.py

@@ -95,7 +95,7 @@ def calculate_channel_hash(channel_key: bytes) -> str:
     return format(hash_bytes[0], "02x")
 
 
-def _decode_path_metadata(path_byte: int) -> tuple[int, int, int]:
+def decode_path_metadata(path_byte: int) -> tuple[int, int, int]:
     """Decode the packed path byte into hop count and byte length."""
     path_hash_size = (path_byte >> 6) + 1
     path_length = path_byte & 0x3F
@@ -135,7 +135,7 @@ def extract_payload(raw_packet: bytes) -> bytes | None:
         # Decode packed path metadata
         if len(raw_packet) < offset + 1:
             return None
-        path_length, _path_hash_size, path_byte_length = _decode_path_metadata(raw_packet[offset])
+        path_length, _path_hash_size, path_byte_length = decode_path_metadata(raw_packet[offset])
         offset += 1
 
         # Skip path data
@@ -171,7 +171,7 @@ def parse_packet(raw_packet: bytes) -> PacketInfo | None:
         # Decode packed path metadata
         if len(raw_packet) < offset + 1:
             return None
-        path_length, path_hash_size, path_byte_length = _decode_path_metadata(raw_packet[offset])
+        path_length, path_hash_size, path_byte_length = decode_path_metadata(raw_packet[offset])
         offset += 1
 
         # Extract path data

app/fanout/apprise_mod.py

@@ -44,8 +44,13 @@ def _format_body(data: dict, *, include_path: bool) -> str:
     via = ""
     if include_path:
         paths = data.get("paths")
-        if paths and isinstance(paths, list) and len(paths) > 0:
-            path_str = paths[0].get("path", "") if isinstance(paths[0], dict) else ""
+        first_path = (
+            paths[0]
+            if isinstance(paths, list) and len(paths) > 0 and isinstance(paths[0], dict)
+            else None
+        )
+        if first_path is not None:
+            path_str = first_path.get("path", "")
         else:
             path_str = None
 
@@ -56,7 +61,7 @@ def _format_body(data: dict, *, include_path: bool) -> str:
             if path_str == "":
                 via = " **via:** [`direct`]"
             else:
-                path_len = paths[0].get("path_len") if isinstance(paths[0], dict) else None
+                path_len = first_path.get("path_len") if first_path is not None else None
                 hop_chars = (
                     len(path_str) // path_len
                     if isinstance(path_len, int) and path_len > 0 and len(path_str) % path_len == 0

app/fanout/community_mqtt.py

@@ -23,6 +23,7 @@ from typing import Any, Protocol
 import aiomqtt
 import nacl.bindings
 
+from app.decoder import decode_path_metadata
 from app.fanout.mqtt_base import BaseMqttPublisher
 
 logger = logging.getLogger(__name__)
@@ -146,16 +147,16 @@ def _calculate_packet_hash(raw_bytes: bytes) -> str:
         if has_transport:
             offset += 4  # Skip 4 bytes of transport codes
 
-        # Read path_len (1 byte on wire). Invalid/truncated packets map to zero hash.
+        # Read packed path metadata. Invalid/truncated packets map to zero hash.
         if offset >= len(raw_bytes):
             return "0" * 16
-        path_len = raw_bytes[offset]
+        path_len, _path_hash_size, path_byte_length = decode_path_metadata(raw_bytes[offset])
         offset += 1
 
         # Skip past path to get to payload. Invalid/truncated packets map to zero hash.
-        if len(raw_bytes) < offset + path_len:
+        if len(raw_bytes) < offset + path_byte_length:
             return "0" * 16
-        payload_start = offset + path_len
+        payload_start = offset + path_byte_length
         payload_data = raw_bytes[payload_start:]
 
         # Hash: payload_type(1 byte) [+ path_len as uint16_t LE for TRACE] + payload_data
@@ -202,20 +203,24 @@ def _decode_packet_fields(raw_bytes: bytes) -> tuple[str, str, str, list[str], i
         if len(raw_bytes) <= offset:
             return route, packet_type, payload_len, path_values, payload_type
 
-        path_len = raw_bytes[offset]
+        path_len, path_hash_size, path_byte_length = decode_path_metadata(raw_bytes[offset])
         offset += 1
 
-        if len(raw_bytes) < offset + path_len:
+        if len(raw_bytes) < offset + path_byte_length:
             return route, packet_type, payload_len, path_values, payload_type
 
-        path_bytes = raw_bytes[offset : offset + path_len]
-        offset += path_len
+        path_bytes = raw_bytes[offset : offset + path_byte_length]
+        offset += path_byte_length
 
         payload_type = (header >> 2) & 0x0F
         route = _ROUTE_MAP.get(route_type, "U")
         packet_type = str(payload_type)
         payload_len = str(max(0, len(raw_bytes) - offset))
-        path_values = [f"{b:02x}" for b in path_bytes]
+        path_values = [
+            path_bytes[i : i + path_hash_size].hex()
+            for i in range(0, len(path_bytes), path_hash_size)
+            if i + path_hash_size <= len(path_bytes)
+        ]
 
         return route, packet_type, payload_len, path_values, payload_type
     except Exception:

app/migrations.py

@@ -13,6 +13,8 @@ from hashlib import sha256
 
 import aiosqlite
 
+from app.decoder import extract_payload, parse_packet
+
 logger = logging.getLogger(__name__)
 
 
@@ -442,35 +444,7 @@ def _extract_payload_for_hash(raw_packet: bytes) -> bytes | None:
 
     Returns the payload bytes, or None if packet is malformed.
     """
-    if len(raw_packet) < 2:
-        return None
-
-    try:
-        header = raw_packet[0]
-        route_type = header & 0x03
-        offset = 1
-
-        # Skip transport codes if present (TRANSPORT_FLOOD=0, TRANSPORT_DIRECT=3)
-        if route_type in (0x00, 0x03):
-            if len(raw_packet) < offset + 4:
-                return None
-            offset += 4
-
-        # Get path length
-        if len(raw_packet) < offset + 1:
-            return None
-        path_length = raw_packet[offset]
-        offset += 1
-
-        # Skip path bytes
-        if len(raw_packet) < offset + path_length:
-            return None
-        offset += path_length
-
-        # Rest is payload (may be empty, matching decoder.py behavior)
-        return raw_packet[offset:]
-    except (IndexError, ValueError):
-        return None
+    return extract_payload(raw_packet)
 
 
 async def _migrate_005_backfill_payload_hashes(conn: aiosqlite.Connection) -> None:
@@ -624,34 +598,10 @@ def _extract_path_from_packet(raw_packet: bytes) -> str | None:
 
     Returns the path as a hex string, or None if packet is malformed.
     """
-    if len(raw_packet) < 2:
-        return None
-
-    try:
-        header = raw_packet[0]
-        route_type = header & 0x03
-        offset = 1
-
-        # Skip transport codes if present (TRANSPORT_FLOOD=0, TRANSPORT_DIRECT=3)
-        if route_type in (0x00, 0x03):
-            if len(raw_packet) < offset + 4:
-                return None
-            offset += 4
-
-        # Get path length
-        if len(raw_packet) < offset + 1:
-            return None
-        path_length = raw_packet[offset]
-        offset += 1
-
-        # Extract path bytes
-        if len(raw_packet) < offset + path_length:
-            return None
-        path_bytes = raw_packet[offset : offset + path_length]
-
-        return path_bytes.hex()
-    except (IndexError, ValueError):
+    packet_info = parse_packet(raw_packet)
+    if packet_info is None:
         return None
+    return packet_info.path.hex()
 
 
 async def _migrate_007_backfill_message_paths(conn: aiosqlite.Connection) -> None:

app/models.py

@@ -102,7 +102,7 @@ class ContactAdvertPath(BaseModel):
     path: str = Field(description="Hex-encoded routing path (empty string for direct)")
     path_len: int = Field(description="Number of hops in the path")
     next_hop: str | None = Field(
-        default=None, description="First hop toward us (2-char hex), or null for direct"
+        default=None, description="First hop toward us, or null for direct"
     )
     first_seen: int = Field(description="Unix timestamp of first observation")
     last_seen: int = Field(description="Unix timestamp of most recent observation")
@@ -201,9 +201,7 @@ class MessagePath(BaseModel):
 
     path: str = Field(description="Hex-encoded routing path")
     received_at: int = Field(description="Unix timestamp when this path was received")
-    path_len: int | None = Field(
-        default=None, description="Number of hops in the path, when known"
-    )
+    path_len: int | None = Field(default=None, description="Number of hops in the path, when known")
 
 
 class Message(BaseModel):

app/packet_processor.py

@@ -91,9 +91,7 @@ async def _handle_duplicate_message(
 
     # Add path if provided
     if path is not None:
-        paths = await MessageRepository.add_path(
-            existing_msg.id, path, received, path_len=path_len
-        )
+        paths = await MessageRepository.add_path(existing_msg.id, path, received, path_len=path_len)
     else:
         # Get current paths for broadcast
         paths = existing_msg.paths or []
@@ -731,6 +729,7 @@ async def _process_advertisement(
         path_hex=new_path_hex,
         timestamp=timestamp,
         max_paths=10,
+        path_len=new_path_len,
     )
 
     # Record name history

app/path_utils.py

@@ -0,0 +1,32 @@
+"""Helpers for working with hex-encoded routing paths."""
+
+
+def get_path_hop_width(path_hex: str | None, path_len: int | None) -> int:
+    """Return hop width in hex chars, falling back to legacy 1-byte hops."""
+    if not path_hex:
+        return 2
+    if isinstance(path_len, int) and path_len > 0 and len(path_hex) % path_len == 0:
+        hop_width = len(path_hex) // path_len
+        if hop_width > 0 and hop_width % 2 == 0:
+            return hop_width
+    return 2
+
+
+def split_path_hops(path_hex: str | None, path_len: int | None) -> list[str]:
+    """Split a hex path string into hop-sized chunks."""
+    if not path_hex:
+        return []
+
+    hop_width = get_path_hop_width(path_hex, path_len)
+    normalized = path_hex.lower()
+    return [
+        normalized[i : i + hop_width]
+        for i in range(0, len(normalized), hop_width)
+        if i + hop_width <= len(normalized)
+    ]
+
+
+def first_path_hop(path_hex: str | None, path_len: int | None) -> str | None:
+    """Return the first hop from a hex path string, if any."""
+    hops = split_path_hops(path_hex, path_len)
+    return hops[0] if hops else None

app/repository/contacts.py

@@ -8,6 +8,7 @@ from app.models import (
     ContactAdvertPathSummary,
     ContactNameHistory,
 )
+from app.path_utils import first_path_hop
 
 
 class AmbiguousPublicKeyPrefixError(ValueError):
@@ -287,7 +288,7 @@ class ContactAdvertPathRepository:
     @staticmethod
     def _row_to_path(row) -> ContactAdvertPath:
         path = row["path_hex"] or ""
-        next_hop = path[:2].lower() if len(path) >= 2 else None
+        next_hop = first_path_hop(path, row["path_len"])
         return ContactAdvertPath(
             path=path,
             path_len=row["path_len"],
@@ -303,6 +304,7 @@ class ContactAdvertPathRepository:
         path_hex: str,
         timestamp: int,
         max_paths: int = 10,
+        path_len: int | None = None,
     ) -> None:
         """
         Upsert a unique advert path observation for a contact and prune to N most recent.
@@ -312,7 +314,7 @@ class ContactAdvertPathRepository:
 
         normalized_key = public_key.lower()
         normalized_path = path_hex.lower()
-        path_len = len(normalized_path) // 2
+        normalized_path_len = path_len if isinstance(path_len, int) else len(normalized_path) // 2
 
         await db.conn.execute(
             """
@@ -324,7 +326,7 @@ class ContactAdvertPathRepository:
                 path_len = excluded.path_len,
                 heard_count = contact_advert_paths.heard_count + 1
             """,
-            (normalized_key, normalized_path, path_len, timestamp, timestamp),
+            (normalized_key, normalized_path, normalized_path_len, timestamp, timestamp),
         )
 
         # Keep only the N most recent unique paths per contact.

app/routers/contacts.py

@@ -16,6 +16,7 @@ from app.models import (
     TraceResponse,
 )
 from app.packet_processor import start_historical_dm_decryption
+from app.path_utils import first_path_hop
 from app.radio import radio_manager
 from app.repository import (
     AmbiguousPublicKeyPrefixError,
@@ -201,11 +202,11 @@ async def get_contact_detail(public_key: str) -> ContactDetail:
         if span_hours > 0:
             advert_frequency = round(total_observations / span_hours, 2)
 
-    # Compute nearest repeaters from first-hop prefixes in advert paths
-    first_hop_stats: dict[str, dict] = {}  # prefix -> {heard_count, path_len, last_seen}
+    # Compute nearest repeaters from first hops in advert paths
+    first_hop_stats: dict[str, dict] = {}  # first hop -> {heard_count, path_len, last_seen}
     for p in advert_paths:
-        if p.path and len(p.path) >= 2:
-            prefix = p.path[:2].lower()
+        prefix = first_path_hop(p.path, p.path_len)
+        if prefix:
             if prefix not in first_hop_stats:
                 first_hop_stats[prefix] = {
                     "heard_count": 0,