Release v0.9.79 runtime and messaging update

Ship the v0.9.79 runtime refresh with transport lane isolation, Infonet secure-message address management, MeshChat MQTT controls, selected asset trail behavior, telemetry panel refinements, onboarding updates, and desktop/package metadata alignment.

Also ignore local graphify work products so analysis folders do not leak into future commits.

backend/services/fetchers/aircraft_database.py

@@ -32,7 +32,7 @@ _REFRESH_INTERVAL_S = 5 * 24 * 3600
 _LIST_TIMEOUT_S = 30
 _DOWNLOAD_TIMEOUT_S = 600
 _USER_AGENT = (
-    "ShadowBroker-OSINT/0.9.75 "
+    "ShadowBroker-OSINT/0.9.79 "
     "(+https://github.com/BigBodyCobain/Shadowbroker; "
     "contact: bigbodycobain@gmail.com)"
 )

backend/services/fetchers/meshtastic_map.py

@@ -182,7 +182,7 @@ def fetch_meshtastic_nodes():
         callsign = str(getattr(get_settings(), "MESHTASTIC_OPERATOR_CALLSIGN", "") or "").strip()
     except Exception:
         callsign = ""
-    ua_base = "ShadowBroker-OSINT/0.9.75 (+https://github.com/BigBodyCobain/Shadowbroker; contact: bigbodycobain@gmail.com; 24h polling)"
+    ua_base = "ShadowBroker-OSINT/0.9.79 (+https://github.com/BigBodyCobain/Shadowbroker; contact: bigbodycobain@gmail.com; 24h polling)"
     user_agent = f"{ua_base}; node={callsign}" if callsign else ua_base
 
     try:

backend/services/fetchers/route_database.py

@@ -25,7 +25,7 @@ _REFRESH_INTERVAL_S = 5 * 24 * 3600
 _HTTP_TIMEOUT_S = 60
 
 _USER_AGENT = (
-    "ShadowBroker-OSINT/0.9.75 "
+    "ShadowBroker-OSINT/0.9.79 "
     "(+https://github.com/BigBodyCobain/Shadowbroker; "
     "contact: bigbodycobain@gmail.com)"
 )

backend/services/mesh/mesh_dm_relay.py

@@ -1264,6 +1264,21 @@ class DMRelay:
                 )
             self._save()
 
+    def unregister_prekey_lookup_alias(self, alias: str) -> bool:
+        """Remove an invite-scoped lookup alias from the local relay."""
+        handle = str(alias or "").strip()
+        if not handle:
+            return False
+        removed = False
+        with self._lock:
+            self._refresh_from_shared_relay()
+            if handle in self._prekey_lookup_aliases:
+                del self._prekey_lookup_aliases[handle]
+                removed = True
+        if removed:
+            self._save()
+        return removed
+
     def consume_one_time_prekey(self, agent_id: str) -> dict[str, Any] | None:
         """Atomically claim the next published one-time prekey for a peer bundle."""
         claimed: dict[str, Any] | None = None

backend/services/mesh/mesh_router.py

@@ -520,7 +520,7 @@ class MeshtasticTransport:
 
             def _on_connect(client, userdata, flags, rc):
                 if rc == 0:
-                    info = client.publish(topic, payload, qos=0)
+                    info = client.publish(topic, payload, qos=1)
                     info.wait_for_publish(timeout=5)
                     published[0] = True
                     client.disconnect()
@@ -550,9 +550,9 @@ class MeshtasticTransport:
                 True,
                 self.NAME,
                 (
-                    f"Published direct to !{to_node:08x} via {region}/{channel}"
+                    f"Broker accepted direct publish to !{to_node:08x} via {region}/{channel}"
                     if direct_node is not None
-                    else f"Published to {region}/{channel} ({len(payload)}B protobuf)"
+                    else f"Broker accepted channel publish to {region}/{channel} ({len(payload)}B protobuf)"
                 ),
             )
         except Exception as e:

backend/services/mesh/mesh_wormhole_identity.py

@@ -11,6 +11,7 @@ import base64
 import hmac
 import hashlib
 import json
+import logging
 import secrets
 import time
 from typing import Any
@@ -51,6 +52,8 @@ PREKEY_LOOKUP_ROTATE_BEFORE_REMAINING_USES = 8
 PREKEY_LOOKUP_ROTATION_OVERLAP_S = 12 * 60 * 60
 PREKEY_LOOKUP_ROTATION_ACTIVE_CAP = 4
 
+logger = logging.getLogger(__name__)
+
 
 def _safe_int(val, default=0) -> int:
     try:
@@ -107,6 +110,7 @@ def _default_identity() -> dict[str, Any]:
 def _prekey_lookup_handle_record(
     handle: str,
     *,
+    label: str = "",
     issued_at: int = 0,
     expires_at: int = 0,
     max_uses: int = 0,
@@ -125,6 +129,7 @@ def _prekey_lookup_handle_record(
     bounded_max_uses = max(1, _safe_int(max_uses or PREKEY_LOOKUP_HANDLE_MAX_USES, PREKEY_LOOKUP_HANDLE_MAX_USES))
     return {
         "handle": str(handle or "").strip(),
+        "label": str(label or "").strip()[:96],
         "issued_at": issued,
         "expires_at": bounded_expires_at,
         "max_uses": bounded_max_uses,
@@ -152,8 +157,10 @@ def _coerce_prekey_lookup_handle_record(
         max_uses = _safe_int(value.get("max_uses", PREKEY_LOOKUP_HANDLE_MAX_USES) or PREKEY_LOOKUP_HANDLE_MAX_USES)
         use_count = _safe_int(value.get("use_count", value.get("uses", 0)) or 0, 0)
         last_used_at = _safe_int(value.get("last_used_at", value.get("last_used", 0)) or 0, 0)
+        label = str(value.get("label", "") or "").strip()
         return _prekey_lookup_handle_record(
             handle,
+            label=label,
             issued_at=issued_at,
             expires_at=expires_at,
             max_uses=max_uses,
@@ -228,6 +235,23 @@ def _fresh_prekey_lookup_handle_record(*, now: int | None = None) -> dict[str, A
     )
 
 
+def _prekey_registration_failure_blocks_dm_invite(detail: str) -> bool:
+    """Only trust-root failures block address export; transport warm-up can finish later."""
+    lowered = str(detail or "").lower()
+    critical_markers = (
+        "root transparency",
+        "external root witness",
+        "stable root",
+        "witness threshold",
+        "witness finality",
+        "root manifest",
+        "root witness",
+        "manifest_fingerprint",
+        "policy fingerprint",
+    )
+    return any(marker in lowered for marker in critical_markers)
+
+
 def _bounded_lookup_handle_records(
     records: list[dict[str, Any]],
     *,
@@ -884,6 +908,7 @@ def export_wormhole_dm_invite(*, label: str = "", expires_in_s: int = 0) -> dict
     existing_handles.append(
         _prekey_lookup_handle_record(
             lookup_handle,
+            label=str(label or "").strip(),
             issued_at=issued_at,
             expires_at=expires_at,
         )
@@ -920,14 +945,25 @@ def export_wormhole_dm_invite(*, label: str = "", expires_in_s: int = 0) -> dict
     except Exception:
         pass
 
+    prekey_registration: dict[str, Any] = {"ok": False, "detail": "prekey bundle publish not attempted"}
     try:
         from services.mesh.mesh_wormhole_prekey import register_wormhole_prekey_bundle
 
-        registered = register_wormhole_prekey_bundle()
-        if not registered.get("ok"):
-            return {"ok": False, "detail": str(registered.get("detail", "") or "prekey bundle registration failed")}
+        prekey_registration = register_wormhole_prekey_bundle()
+        if not prekey_registration.get("ok"):
+            detail = str(prekey_registration.get("detail", "") or "prekey bundle registration failed")
+            if _prekey_registration_failure_blocks_dm_invite(detail):
+                return {"ok": False, "detail": detail}
+            logger.warning(
+                "DM invite prekey publish pending: %s",
+                detail,
+            )
     except Exception as exc:
-        return {"ok": False, "detail": str(exc) or "prekey bundle registration failed"}
+        prekey_registration = {"ok": False, "detail": str(exc) or "prekey bundle registration failed"}
+        detail = str(prekey_registration.get("detail", "") or "")
+        if _prekey_registration_failure_blocks_dm_invite(detail):
+            return {"ok": False, "detail": detail}
+        logger.warning("DM invite prekey publish pending: %s", prekey_registration["detail"])
 
     invite_node_id, invite_public_key, invite_private_key = _generate_invite_signing_identity()
     payload = _attach_dm_invite_root_distribution(payload)
@@ -958,6 +994,8 @@ def export_wormhole_dm_invite(*, label: str = "", expires_in_s: int = 0) -> dict
         "peer_id": str(invite_node_id or ""),
         "trust_fingerprint": str(payload.get("identity_commitment", "") or ""),
         "invite": invite,
+        "prekey_publish_pending": not bool(prekey_registration.get("ok")),
+        "prekey_registration": prekey_registration,
     }
 
 
@@ -980,6 +1018,140 @@ def get_prekey_lookup_handle_records() -> list[dict[str, Any]]:
     ]
 
 
+def list_prekey_lookup_handle_records_for_ui(*, now: int | None = None) -> dict[str, Any]:
+    """Return shareable DM address records without exposing local identity secrets."""
+    current_time = _safe_int(now or time.time(), int(time.time()))
+    addresses: list[dict[str, Any]] = []
+    for record in get_prekey_lookup_handle_records():
+        handle = str(record.get("handle", "") or "").strip()
+        if not handle:
+            continue
+        expires_at = _effective_prekey_lookup_handle_expires_at(record)
+        max_uses = max(
+            1,
+            _safe_int(
+                record.get("max_uses", PREKEY_LOOKUP_HANDLE_MAX_USES) or PREKEY_LOOKUP_HANDLE_MAX_USES,
+                PREKEY_LOOKUP_HANDLE_MAX_USES,
+            ),
+        )
+        use_count = max(0, _safe_int(record.get("use_count", 0) or 0, 0))
+        addresses.append(
+            {
+                "handle": handle,
+                "label": str(record.get("label", "") or "").strip(),
+                "issued_at": _safe_int(record.get("issued_at", 0) or 0, 0),
+                "expires_at": expires_at,
+                "max_uses": max_uses,
+                "use_count": use_count,
+                "remaining_uses": max(0, max_uses - use_count),
+                "last_used_at": _safe_int(record.get("last_used_at", 0) or 0, 0),
+                "expired": bool(expires_at > 0 and current_time >= expires_at),
+                "exhausted": bool(use_count >= max_uses),
+            }
+        )
+    addresses.sort(key=lambda item: _safe_int(item.get("issued_at", 0) or 0, 0), reverse=True)
+    return {"ok": True, "addresses": addresses}
+
+
+def rename_prekey_lookup_handle(handle: str, label: str) -> dict[str, Any]:
+    """Rename an active invite-scoped DM lookup handle without changing the handle."""
+    lookup_handle = str(handle or "").strip()
+    next_label = str(label or "").strip()[:96]
+    if not lookup_handle:
+        return {"ok": False, "detail": "missing_lookup_handle"}
+
+    current_time = int(time.time())
+    data = read_wormhole_identity()
+    existing, _ = _normalize_prekey_lookup_handles(
+        data.get("prekey_lookup_handles", []),
+        fallback_issued_at=current_time,
+        now=current_time,
+    )
+    updated = False
+    next_records: list[dict[str, Any]] = []
+    for record in existing:
+        current = dict(record)
+        if str(current.get("handle", "") or "").strip() == lookup_handle:
+            current["label"] = next_label
+            updated = True
+        next_records.append(current)
+
+    if not updated:
+        return {
+            "ok": False,
+            "handle": lookup_handle,
+            "label": next_label,
+            "updated": False,
+            "detail": "lookup_handle_not_found",
+        }
+
+    normalized_records, _ = _normalize_prekey_lookup_handles(
+        next_records,
+        fallback_issued_at=current_time,
+        now=current_time,
+    )
+    _write_identity({"prekey_lookup_handles": normalized_records})
+    return {
+        "ok": True,
+        "handle": lookup_handle,
+        "label": next_label,
+        "updated": True,
+    }
+
+
+def revoke_prekey_lookup_handle(handle: str) -> dict[str, Any]:
+    """Revoke an invite-scoped DM lookup handle for future first-contact attempts."""
+    lookup_handle = str(handle or "").strip()
+    if not lookup_handle:
+        return {"ok": False, "detail": "missing_lookup_handle"}
+    current_time = int(time.time())
+    data = read_wormhole_identity()
+    existing, _ = _normalize_prekey_lookup_handles(
+        data.get("prekey_lookup_handles", []),
+        fallback_issued_at=current_time,
+        now=current_time,
+    )
+    next_records = [
+        dict(record)
+        for record in existing
+        if str(record.get("handle", "") or "").strip() != lookup_handle
+    ]
+    identity_removed = len(next_records) != len(existing)
+    if identity_removed:
+        _write_identity({"prekey_lookup_handles": next_records})
+
+    relay_removed = False
+    try:
+        from services.mesh.mesh_dm_relay import dm_relay
+
+        relay_removed = bool(dm_relay.unregister_prekey_lookup_alias(lookup_handle))
+    except Exception:
+        relay_removed = False
+
+    republished = False
+    detail = ""
+    if identity_removed:
+        try:
+            from services.mesh.mesh_wormhole_prekey import register_wormhole_prekey_bundle
+
+            registered = register_wormhole_prekey_bundle()
+            republished = bool(registered.get("ok"))
+            if not republished:
+                detail = str(registered.get("detail", "") or "prekey bundle republish failed")
+        except Exception as exc:
+            detail = str(exc) or "prekey bundle republish failed"
+
+    return {
+        "ok": True,
+        "handle": lookup_handle,
+        "revoked": bool(identity_removed or relay_removed),
+        "identity_removed": identity_removed,
+        "relay_removed": relay_removed,
+        "republished": republished,
+        "detail": detail,
+    }
+
+
 def record_prekey_lookup_handle_use(handle: str, *, now: int | None = None) -> dict[str, Any] | None:
     lookup_handle = str(handle or "").strip()
     if not lookup_handle:
@@ -999,6 +1171,7 @@ def record_prekey_lookup_handle_use(handle: str, *, now: int | None = None) -> d
         if str(current.get("handle", "") or "").strip() == lookup_handle:
             current = _prekey_lookup_handle_record(
                 lookup_handle,
+                label=str(current.get("label", "") or "").strip(),
                 issued_at=_safe_int(current.get("issued_at", 0) or 0, current_time),
                 expires_at=_safe_int(current.get("expires_at", 0) or 0, 0),
                 max_uses=_safe_int(current.get("max_uses", PREKEY_LOOKUP_HANDLE_MAX_USES) or PREKEY_LOOKUP_HANDLE_MAX_USES),
@@ -1129,6 +1302,7 @@ def maybe_rotate_prekey_lookup_handles(*, now: int | None = None) -> dict[str, A
         candidate_records.append(
             _prekey_lookup_handle_record(
                 old_handle,
+                label=str(record.get("label", "") or "").strip(),
                 issued_at=_safe_int(record.get("issued_at", 0) or 0, current_time),
                 expires_at=overlap_expires_at,
                 max_uses=_safe_int(record.get("max_uses", PREKEY_LOOKUP_HANDLE_MAX_USES) or PREKEY_LOOKUP_HANDLE_MAX_USES),

backend/services/mesh/mesh_wormhole_root_manifest.py

@@ -12,6 +12,7 @@ from __future__ import annotations
 import base64
 import hashlib
 import json
+import logging
 import time
 from pathlib import Path
 from typing import Any
@@ -23,7 +24,7 @@ from cryptography.hazmat.primitives.asymmetric import ed25519
 
 from services.mesh.mesh_crypto import build_signature_payload, derive_node_id, verify_node_binding, verify_signature
 from services.mesh.mesh_protocol import PROTOCOL_VERSION
-from services.mesh.mesh_secure_storage import read_domain_json, write_domain_json
+from services.mesh.mesh_secure_storage import SecureStorageError, read_domain_json, write_domain_json
 from services.mesh.mesh_wormhole_identity import root_identity_fingerprint_for_material
 from services.mesh.mesh_wormhole_persona import (
     bootstrap_wormhole_persona_state,
@@ -51,6 +52,7 @@ DEFAULT_ROOT_WITNESS_THRESHOLD = 2
 DEFAULT_ROOT_WITNESS_MANAGEMENT_SCOPE = "local"
 DEFAULT_ROOT_WITNESS_INDEPENDENCE_GROUP = "local_system"
 DEFAULT_ROOT_EXTERNAL_WITNESS_MAX_AGE_S = 3600
+logger = logging.getLogger(__name__)
 
 
 def _safe_int(val: Any, default: int = 0) -> int:
@@ -461,12 +463,22 @@ def witness_policy_fingerprint(policy: dict[str, Any]) -> str:
 
 
 def read_root_distribution_state() -> dict[str, Any]:
-    raw = read_domain_json(
-        ROOT_DISTRIBUTION_DOMAIN,
-        ROOT_DISTRIBUTION_FILE,
-        _default_state,
-        base_dir=DATA_DIR,
-    )
+    try:
+        raw = read_domain_json(
+            ROOT_DISTRIBUTION_DOMAIN,
+            ROOT_DISTRIBUTION_FILE,
+            _default_state,
+            base_dir=DATA_DIR,
+        )
+    except SecureStorageError as exc:
+        detail = str(exc)
+        if "Failed to decrypt domain JSON" not in detail:
+            raise
+        logger.warning(
+            "Root distribution state could not decrypt; regenerating local witness distribution: %s",
+            detail,
+        )
+        raw = _default_state()
     state = {**_default_state(), **dict(raw or {})}
     state["witness_identity"] = {**_empty_witness_identity(), **dict(state.get("witness_identity") or {})}
     witness_identities, witness_changed = _normalize_witness_identities(

backend/services/mesh/meshtastic_topics.py

@@ -108,8 +108,18 @@ def normalize_topic_filter(value: str) -> str | None:
     return "/".join(parts)
 
 
-def _default_topic_for_root(root: str) -> str:
-    return f"msh/{root}/2/e/{DEFAULT_CHANNEL}/#"
+def _default_topics_for_root(root: str) -> list[str]:
+    """Return the default LongFast subscriptions for a region root.
+
+    The public broker carries protobuf/encrypted traffic under ``/e/`` and
+    companion decoded JSON traffic under ``/json/``. Positions often arrive on
+    the protobuf path, while public text is commonly easiest to observe on the
+    JSON path.
+    """
+    return [
+        f"msh/{root}/2/e/{DEFAULT_CHANNEL}/#",
+        f"msh/{root}/2/json/{DEFAULT_CHANNEL}/#",
+    ]
 
 
 def build_subscription_topics(
@@ -124,7 +134,11 @@ def build_subscription_topics(
         # via MESH_MQTT_EXTRA_ROOTS to avoid flooding the public broker.
     roots.extend(root for root in (normalize_root(item) for item in _split_config_values(extra_roots)) if root)
 
-    topics = [_default_topic_for_root(root) for root in _dedupe(roots)]
+    topics = [
+        topic
+        for root in _dedupe(roots)
+        for topic in _default_topics_for_root(root)
+    ]
     topics.extend(
         topic
         for topic in (

backend/services/network_utils.py

@@ -73,7 +73,7 @@ def fetch_with_curl(url, method="GET", json_data=None, timeout=15, headers=None,
     both Python requests and the barebones Windows system curl.
     """
     default_headers = {
-        "User-Agent": "ShadowBroker-OSINT/0.9.75 (+https://github.com/BigBodyCobain/Shadowbroker; contact: bigbodycobain@gmail.com)",
+        "User-Agent": "ShadowBroker-OSINT/0.9.79 (+https://github.com/BigBodyCobain/Shadowbroker; contact: bigbodycobain@gmail.com)",
     }
     if headers:
         default_headers.update(headers)

backend/services/shodan_connector.py

@@ -20,7 +20,7 @@ from cachetools import TTLCache
 logger = logging.getLogger(__name__)
 
 _SHODAN_BASE = "https://api.shodan.io"
-_USER_AGENT = "ShadowBroker/0.9.75 local Shodan connector"
+_USER_AGENT = "ShadowBroker/0.9.79 local Shodan connector"
 _REQUEST_TIMEOUT = 15
 _MIN_INTERVAL_SECONDS = 1.05  # Shodan docs say API plans are rate limited to ~1 req/sec.
 _DEFAULT_SEARCH_PAGES = 1

backend/services/sigint_bridge.py

@@ -545,6 +545,198 @@ class MeshtasticBridge:
         self._message_dedupe[key] = now
         return False
 
+    @staticmethod
+    def _message_dedupe_key(message: dict) -> str:
+        sender = str(message.get("from") or "???").strip().lower()
+        recipient = str(message.get("to") or "broadcast").strip().lower()
+        text = str(message.get("text") or "").strip()
+        channel = str(message.get("channel") or "LongFast").strip().lower()
+        root = str(message.get("root") or message.get("region") or "").strip().lower()
+        if root == "us":
+            root = "us"
+        return f"{sender}:{recipient}:{root}:{channel}:{text}"
+
+    def append_text_message(self, message: dict, *, dedupe_window_s: float = 5.0) -> bool:
+        """Append a Meshtastic text message unless it is a near-immediate echo."""
+        if not str(message.get("text") or "").strip():
+            return False
+        now = time.time()
+        cutoff = now - max(1.0, dedupe_window_s)
+        next_message = dict(message)
+        next_message.setdefault("to", "broadcast")
+        next_message.setdefault("channel", "LongFast")
+        next_message.setdefault("timestamp", datetime.utcnow().isoformat() + "Z")
+        key = self._message_dedupe_key(next_message)
+        for existing in list(self.messages)[:40]:
+            if self._message_dedupe_key(existing) != key:
+                continue
+            try:
+                existing_ts_raw = existing.get("timestamp")
+                existing_ts = (
+                    datetime.fromisoformat(str(existing_ts_raw).replace("Z", "+00:00")).timestamp()
+                    if existing_ts_raw
+                    else now
+                )
+            except Exception:
+                existing_ts = now
+            if existing_ts >= cutoff:
+                if not existing.get("root") and next_message.get("root"):
+                    existing["root"] = next_message.get("root")
+                if not existing.get("region") and next_message.get("region"):
+                    existing["region"] = next_message.get("region")
+                return False
+        self.messages.appendleft(next_message)
+        return True
+
+    @staticmethod
+    def _coerce_node_ref(value) -> str:
+        """Normalize Meshtastic node identifiers into the public !xxxxxxxx form."""
+        if value is None:
+            return ""
+        if isinstance(value, int):
+            return f"!{value & 0xFFFFFFFF:08x}"
+        raw = str(value).strip()
+        if not raw:
+            return ""
+        if raw.startswith("!"):
+            return raw
+        lowered = raw.lower()
+        if lowered.startswith("0x"):
+            try:
+                return f"!{int(lowered, 16) & 0xFFFFFFFF:08x}"
+            except ValueError:
+                return raw
+        if raw.isdigit():
+            try:
+                return f"!{int(raw) & 0xFFFFFFFF:08x}"
+            except ValueError:
+                return raw
+        if len(raw) == 8 and all(ch in "0123456789abcdefABCDEF" for ch in raw):
+            return f"!{raw.lower()}"
+        return raw
+
+    @staticmethod
+    def _first_text_value(*values) -> str:
+        for value in values:
+            if isinstance(value, bytes):
+                value = value.decode("utf-8", errors="replace")
+            if isinstance(value, str):
+                text = value.strip()
+                if text:
+                    return MeshtasticBridge._repair_text_mojibake(text)
+        return ""
+
+    @staticmethod
+    def _repair_text_mojibake(text: str) -> str:
+        """Repair common UTF-8-as-Latin-1 mojibake from MQTT JSON bridges."""
+        if not text or not any(marker in text for marker in ("Ã", "Ð", "Ñ")):
+            return text
+        try:
+            repaired = text.encode("latin-1").decode("utf-8").strip()
+        except UnicodeError:
+            return text
+        if repaired and repaired != text:
+            return repaired
+        return text
+
+    @staticmethod
+    def _first_present(*values):
+        for value in values:
+            if value is not None and value != "":
+                return value
+        return None
+
+    def _extract_json_text_message(self, data: dict, topic: str) -> dict | None:
+        """Extract a public Meshtastic text event from decoded MQTT JSON.
+
+        Meshtastic JSON brokers are not perfectly uniform. Some packets expose
+        text at the top level, some under ``decoded`` or ``payload``. Keep this
+        permissive for receive, but only return messages with non-empty text.
+        """
+        if not isinstance(data, dict):
+            return None
+        topic_meta = parse_topic_metadata(topic)
+        packet = data.get("packet") if isinstance(data.get("packet"), dict) else {}
+        decoded = data.get("decoded") if isinstance(data.get("decoded"), dict) else {}
+        payload_obj = data.get("payload")
+        payload = payload_obj if isinstance(payload_obj, dict) else {}
+        decoded_payload_obj = decoded.get("payload") if decoded else None
+        decoded_payload = decoded_payload_obj if isinstance(decoded_payload_obj, dict) else {}
+
+        text = self._first_text_value(
+            data.get("text"),
+            data.get("message"),
+            data.get("msg"),
+            payload_obj if isinstance(payload_obj, str) else "",
+            payload.get("text"),
+            payload.get("message"),
+            payload.get("msg"),
+            payload.get("payload") if isinstance(payload.get("payload"), str) else "",
+            decoded.get("text"),
+            decoded.get("message"),
+            decoded.get("payload") if isinstance(decoded.get("payload"), str) else "",
+            decoded_payload.get("text"),
+            decoded_payload.get("message"),
+            decoded_payload.get("msg"),
+        )
+        if not text:
+            return None
+
+        sender = self._coerce_node_ref(
+            self._first_present(
+                data.get("from"),
+                data.get("fromId"),
+                data.get("from_id"),
+                data.get("sender"),
+                data.get("senderId"),
+                data.get("sender_id"),
+                packet.get("from"),
+                packet.get("fromId"),
+                packet.get("from_id"),
+                decoded.get("from"),
+            )
+        )
+        recipient = self._coerce_node_ref(
+            self._first_present(
+                data.get("to"),
+                data.get("toId"),
+                data.get("to_id"),
+                data.get("recipient"),
+                data.get("recipientId"),
+                data.get("recipient_id"),
+                packet.get("to"),
+                packet.get("toId"),
+                packet.get("to_id"),
+                decoded.get("to"),
+            )
+        )
+        if not recipient or recipient in {"!ffffffff", "broadcast"}:
+            recipient = "broadcast"
+
+        timestamp = datetime.utcnow().isoformat() + "Z"
+        rx_time = self._first_present(
+            data.get("rxTime"),
+            data.get("rx_time"),
+            data.get("timestamp"),
+            packet.get("rxTime"),
+            packet.get("timestamp"),
+        )
+        if isinstance(rx_time, (int, float)) and rx_time > 0:
+            try:
+                timestamp = datetime.fromtimestamp(float(rx_time), tz=timezone.utc).isoformat()
+            except (OSError, ValueError):
+                pass
+
+        return {
+            "from": sender or topic.split("/")[-1],
+            "to": recipient,
+            "text": text[:500],
+            "region": topic_meta["region"],
+            "root": topic_meta["root"],
+            "channel": topic_meta["channel"],
+            "timestamp": timestamp,
+        }
+
     def start(self):
         if self._thread and self._thread.is_alive():
             if not self._stop.is_set():
@@ -693,6 +885,9 @@ class MeshtasticBridge:
             if "/json/" in topic:
                 try:
                     data = json.loads(payload)
+                    text_message = self._extract_json_text_message(data, topic)
+                    if text_message:
+                        self.append_text_message(text_message, dedupe_window_s=30.0)
                     if self._rate_limited():
                         return
                     self._ingest_data(data, topic)
@@ -715,7 +910,7 @@ class MeshtasticBridge:
                         topic_meta["root"],
                     ):
                         return
-                    self.messages.appendleft(
+                    self.append_text_message(
                         {
                             "from": data.get("from", "???"),
                             "to": recipient,

backend/services/transport_lane_isolation.py

@@ -0,0 +1,57 @@
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+
+def disable_public_mesh_lane(*, reason: str = "private_lane_enabled") -> dict[str, Any]:
+    """Disable public Meshtastic MQTT before private Wormhole/Infonet starts."""
+    result: dict[str, Any] = {
+        "ok": True,
+        "reason": reason,
+        "settings_disabled": False,
+        "runtime_stopped": False,
+    }
+
+    # Scheduled Wormhole prewarm must not mutate the user's explicit public
+    # MeshChat session. Only a deliberate private-lane activation should sever
+    # the public MQTT lane.
+    normalized_reason = str(reason or "").strip().lower()
+    if normalized_reason == "wormhole_scheduled_prewarm" or normalized_reason.endswith(":scheduled_prewarm"):
+        try:
+            from services.meshtastic_mqtt_settings import mqtt_bridge_enabled
+
+            if mqtt_bridge_enabled():
+                logger.info("Keeping public Mesh lane active during Wormhole prewarm: %s", reason)
+                result["skipped"] = True
+                result["skip_reason"] = "public_mesh_user_enabled"
+                return result
+        except Exception as exc:
+            logger.debug("Could not inspect public Mesh state during %s: %s", reason, exc)
+
+    logger.info("Disabling public Mesh lane: %s", reason)
+
+    try:
+        from services.meshtastic_mqtt_settings import write_meshtastic_mqtt_settings
+
+        settings = write_meshtastic_mqtt_settings(enabled=False)
+        result["settings_disabled"] = not bool(settings.get("enabled"))
+    except Exception as exc:
+        logger.warning("Failed to disable public Mesh settings during %s: %s", reason, exc)
+        result["ok"] = False
+        result["settings_error"] = str(exc)
+
+    try:
+        from services.sigint_bridge import sigint_grid
+
+        if sigint_grid.mesh.is_running():
+            sigint_grid.mesh.stop()
+        result["runtime_stopped"] = not sigint_grid.mesh.is_running()
+    except Exception as exc:
+        logger.warning("Failed to stop public Mesh runtime during %s: %s", reason, exc)
+        result["ok"] = False
+        result["runtime_error"] = str(exc)
+
+    return result

backend/services/unusual_whales_connector.py

@@ -24,7 +24,7 @@ from cachetools import TTLCache
 logger = logging.getLogger(__name__)
 
 _FINNHUB_BASE = "https://finnhub.io/api/v1"
-_USER_AGENT = "ShadowBroker/0.9.75 Finnhub connector"
+_USER_AGENT = "ShadowBroker/0.9.79 Finnhub connector"
 _REQUEST_TIMEOUT = 12
 _MIN_INTERVAL_SECONDS = 0.35  # Stay well under 60 calls/min
 

backend/services/wormhole_supervisor.py

@@ -243,6 +243,48 @@ def _pid_alive(pid: int) -> bool:
     return True
 
 
+def _find_wormhole_server_pid() -> int:
+    if os.name == "nt":
+        return 0
+    proc_dir = Path("/proc")
+    if not proc_dir.exists():
+        return 0
+    current_pid = os.getpid()
+    script_name = WORMHOLE_SCRIPT.name
+    script_path = str(WORMHOLE_SCRIPT)
+    for entry in proc_dir.iterdir():
+        if not entry.name.isdigit():
+            continue
+        pid = int(entry.name)
+        if pid == current_pid:
+            continue
+        try:
+            raw = (entry / "cmdline").read_bytes()
+        except OSError:
+            continue
+        cmdline = raw.replace(b"\x00", b" ").decode("utf-8", errors="replace")
+        if script_path in cmdline or script_name in cmdline:
+            return pid
+    return 0
+
+
+def _terminate_pid(pid: int, *, timeout_s: float = 5.0) -> None:
+    if os.name == "nt" or pid <= 0:
+        return
+    try:
+        os.kill(pid, signal.SIGTERM)
+    except Exception:
+        return
+    deadline = time.monotonic() + timeout_s
+    while time.monotonic() < deadline and _pid_alive(pid):
+        time.sleep(0.1)
+    if _pid_alive(pid):
+        try:
+            os.kill(pid, signal.SIGKILL)
+        except Exception:
+            pass
+
+
 def _probe_ready(timeout_s: float = 1.5) -> bool:
     try:
         with urlopen(f"http://{WORMHOLE_HOST}:{WORMHOLE_PORT}/api/health", timeout=timeout_s) as resp:
@@ -266,17 +308,32 @@ def _probe_json(path: str, timeout_s: float = 1.5) -> dict[str, Any] | None:
 def _current_runtime_state() -> dict[str, Any]:
     settings = read_wormhole_settings()
     status = read_wormhole_status()
+    configured = bool(settings.get("enabled"))
     running = False
+    ready = False
     pid = int(status.get("pid", 0) or 0)
-    if _PROCESS and _PROCESS.poll() is None:
+    if not configured:
+        # Disabled private transport must stay disabled even if a stale local
+        # wormhole process is still answering on the health port. Public
+        # MeshChat relies on this state to keep the MQTT and Wormhole lanes
+        # mutually exclusive.
+        pid = 0
+        ready = False
+    elif _PROCESS and _PROCESS.poll() is None:
         running = True
         pid = int(_PROCESS.pid or 0)
-    elif _pid_alive(pid):
-        running = True
-    elif _probe_ready(timeout_s=0.35):
-        running = True
-        pid = 0
-    ready = running and _probe_ready()
+    else:
+        if _pid_alive(pid):
+            running = True
+        else:
+            discovered_pid = _find_wormhole_server_pid()
+            if discovered_pid > 0:
+                running = True
+                pid = discovered_pid
+        if not running and _probe_ready(timeout_s=0.35):
+            running = True
+            pid = 0
+        ready = running and _probe_ready()
     if not running:
         pid = 0
     transport_active = status.get("transport_active", "") if ready else ""
@@ -319,13 +376,13 @@ def _current_runtime_state() -> dict[str, Any]:
     anonymous_mode = bool(settings.get("anonymous_mode"))
     anonymous_mode_ready = bool(
         anonymous_mode
-        and settings.get("enabled")
+        and configured
         and ready
         and effective_transport in {"tor", "tor_arti", "i2p", "mixnet"}
     )
     snapshot = {
         "installed": _installed(),
-        "configured": bool(settings.get("enabled")),
+        "configured": configured,
         "running": running,
         "ready": ready,
         "transport_configured": str(settings.get("transport", "direct") or "direct"),
@@ -395,6 +452,12 @@ def get_wormhole_state() -> dict[str, Any]:
 def connect_wormhole(*, reason: str = "connect") -> dict[str, Any]:
     with _LOCK:
         _invalidate_state_cache()
+        try:
+            from services.transport_lane_isolation import disable_public_mesh_lane
+
+            disable_public_mesh_lane(reason=f"wormhole_{reason}")
+        except Exception as exc:
+            logger.warning("Failed to enforce public/private lane isolation during %s: %s", reason, exc)
         settings = read_wormhole_settings()
         if not settings.get("enabled"):
             settings = settings.copy()
@@ -487,8 +550,8 @@ def connect_wormhole(*, reason: str = "connect") -> dict[str, Any]:
 def disconnect_wormhole(*, reason: str = "disconnect") -> dict[str, Any]:
     with _LOCK:
         _invalidate_state_cache()
-        current = _current_runtime_state()
-        pid = int(current.get("pid", 0) or 0)
+        status = read_wormhole_status()
+        pid = int(status.get("pid", 0) or 0)
         global _PROCESS
         if _PROCESS and _PROCESS.poll() is None:
             try:
@@ -499,14 +562,15 @@ def disconnect_wormhole(*, reason: str = "disconnect") -> dict[str, Any]:
                     _PROCESS.kill()
                 except Exception:
                     pass
-        elif os.name != "nt" and _pid_alive(pid):
-            try:
-                os.kill(pid, signal.SIGTERM)
-            except Exception:
-                pass
+        if os.name != "nt":
+            _terminate_pid(pid)
+            discovered_pid = _find_wormhole_server_pid()
+            if discovered_pid > 0 and discovered_pid != pid:
+                _terminate_pid(discovered_pid)
         _PROCESS = None
         write_wormhole_status(
             reason=reason,
+            configured=False,
             running=False,
             ready=False,
             pid=0,