chore(release): update v0.9.81 SHA256 digests after rebuild

Re-cut v0.9.81 binaries from current main (which now includes the
private gate + DM hashchain spool from #326 and the gate-directory
test from #327). All three artifacts were signed with the same
minisign updater key as the original v0.9.81 release, so existing
v0.9.81 installs on Tauri auto-update accept the new bundles.

Updated hashes (verified against released assets):
- ShadowBroker_v0.9.81.zip      f81f454bdc88e9a32c351df38212b8cfa624704d65764b971bb091eef62259c6
- ShadowBroker_0.9.81_x64-setup.exe   25e9a95d0d8ce959a7d08fe8e7406772ae24b596652793e81d1de5d02510a5a6
- ShadowBroker_0.9.81_x64_en-US.msi   34e655fc0c0f195ee4ac978f228a4b2b9d5565253b8771aca9ef4693409e9e70

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

README.md

@@ -174,7 +174,7 @@ ShadowBroker v0.9.7 ships **InfoNet** (decentralized intelligence mesh + Soverei
 | Channel | Privacy Status | Details |
 |---|---|---|
 | **Meshtastic / APRS** | **PUBLIC** | RF radio transmissions are public and interceptable by design. |
-| **InfoNet Gate Chat** | **OBFUSCATED** | Messages are obfuscated with gate personas and canonical payload signing, but NOT end-to-end encrypted. Metadata is not hidden. |
+| **InfoNet Gate Chat** | **OBFUSCATED** | Messages are obfuscated with gate personas and canonical payload signing, but NOT end-to-end encrypted. Metadata is not hidden despite being designed through Tor and Reticulum (Work in progress). |
 | **Dead Drop DMs** | **STRONGEST CURRENT LANE** | Token-based epoch mailbox with SAS word verification. Strongest lane in this build, but not yet confidently private. |
 | **Sovereign Shell governance** | **PUBLIC LEDGER** | Petitions, votes, upgrade hashes, and dispute stakes are signed events on a public hashchain. Pseudonymous via gate persona, but governance actions are intentionally observable. |
 | **Privacy primitives (RingCT / stealth / DEX)** | **NOT YET WIRED** | Locked Protocol contracts are in place, but the cryptographic scheme has not been chosen. The privacy-core Rust crate is the integration target for a future sprint. |
@@ -199,7 +199,7 @@ The first decentralized intelligence communication and governance layer built di
 
 **Communication layer (since v0.9.6):**
 
-* **InfoNet Experimental Testnet** — A global, obfuscated message relay. Anyone running ShadowBroker can transmit and receive on the InfoNet. Messages pass through a Wormhole relay layer with gate personas, Ed25519 canonical payload signing, and transport obfuscation.
+* **InfoNet Experimental Testnet** — A global, obfuscated message relay using Tor and Reticulum. Anyone running ShadowBroker can transmit and receive on the InfoNet. Messages pass through a Wormhole relay layer with gate personas, Ed25519 canonical payload signing, and transport obfuscation.
 * **Mesh Chat Panel** — Three-tab interface: **INFONET** (gate chat with obfuscated transport), **MESH** (Meshtastic radio integration), **DEAD DROP** (peer-to-peer message exchange with token-based epoch mailboxes — strongest current lane).
 * **Gate Persona System** — Pseudonymous identities with Ed25519 signing keys, prekey bundles, SAS word contact verification, and abuse reporting.
 * **Mesh Terminal** — Built-in CLI: `send`, `dm`, market commands, gate state inspection. Draggable panel, minimizes to the top bar. Type `help` to see all commands.

backend/.env.example

@@ -11,6 +11,13 @@ AIS_API_KEY=              # https://aisstream.io/ — free tier WebSocket key
 
 # ── Optional ───────────────────────────────────────────────────
 
+# AISHub REST fallback. Used when stream.aisstream.io is unreachable
+# (e.g. their cert expires or server goes offline). Free tier requires
+# registration at https://www.aishub.net/api. Poll cadence defaults to
+# 20 min to stay courteous; tunable via AISHUB_POLL_INTERVAL_MINUTES.
+# AISHUB_USERNAME=
+# AISHUB_POLL_INTERVAL_MINUTES=20
+
 # Override allowed CORS origins (comma-separated). Defaults to localhost + LAN auto-detect.
 # CORS_ORIGINS=http://192.168.1.50:3000,https://my-domain.com
 

backend/data/release_digests.json

@@ -36,5 +36,15 @@
     "ShadowBroker_v0.9.79.zip": "f6877c1d66614525315ea82636ce9f7b41178332c4dbf90d27431a1ea1d9cd47",
     "ShadowBroker_0.9.79_x64-setup.exe": "f7b676ada45cac7da05868b0a353678c9ee700e3abcf456a7c0c038c36da446f",
     "ShadowBroker_0.9.79_x64_en-US.msi": "e0713c3cdda184cfbea750bfac0d62a35678fec00847e6476f2cac8e7e42046e"
+  },
+  "v0.9.8": {
+    "ShadowBroker_v0.9.8.zip": "183bb5cd62b9b9349d95df5ef7696cb6ca810ab4b991fa9dab6f898af4c7a175",
+    "ShadowBroker_0.9.8_x64-setup.exe": "94a0309862e9c81c92cdcbfea8eec9dbb97eef19ded82b26217b397defbc810c",
+    "ShadowBroker_0.9.8_x64_en-US.msi": "fe22f9d51e4360d74c18a7250c2fbb9ed4fa4c7a884b3ac0d04a21115466386b"
+  },
+  "v0.9.81": {
+    "ShadowBroker_v0.9.81.zip": "f81f454bdc88e9a32c351df38212b8cfa624704d65764b971bb091eef62259c6",
+    "ShadowBroker_0.9.81_x64-setup.exe": "25e9a95d0d8ce959a7d08fe8e7406772ae24b596652793e81d1de5d02510a5a6",
+    "ShadowBroker_0.9.81_x64_en-US.msi": "34e655fc0c0f195ee4ac978f228a4b2b9d5565253b8771aca9ef4693409e9e70"
   }
 }

backend/main.py

@@ -1,4 +1,4 @@
-import os
+import os
 from dotenv import load_dotenv
 load_dotenv()
 
@@ -8,18 +8,24 @@ import asyncio
 import base64
 import hmac
 import importlib
+import ipaddress
 import secrets
 import hashlib as _hashlib_mod
 from dataclasses import dataclass, field
 from typing import Any
 from json import JSONDecodeError
 
-APP_VERSION = "0.9.79"
+APP_VERSION = "0.9.81"
 
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
 _start_time = time.time()
 _MESH_ONLY = os.environ.get("MESH_ONLY", "").strip().lower() in ("1", "true", "yes")
+_HEADLESS_MESH_NODE_RUNTIME = os.environ.get("SHADOWBROKER_MESH_NODE_RUNTIME", "").strip().lower() in (
+    "1",
+    "true",
+    "yes",
+)
 _WARNED_LEGACY_DM_PUBKEY_LOOKUPS: set[str] = set()
 
 
@@ -1095,6 +1101,7 @@ _WORMHOLE_PUBLIC_PROFILE_FIELDS = {"wormhole_enabled"}
 _PRIVATE_LANE_CONTROL_FIELDS = {"private_lane_tier", "private_lane_policy"}
 _PUBLIC_RNS_STATUS_FIELDS = {"enabled", "ready", "configured_peers", "active_peers"}
 _NODE_PUBLIC_EVENT_HOOK_REGISTERED = False
+_NODE_RUNTIME_THREADS_STARTED = False
 _INFONET_PRIVATE_TRANSPORT_LOCK = threading.Lock()
 
 
@@ -1184,6 +1191,49 @@ def _filter_infonet_sync_records(records: list[Any]) -> list[Any]:
     ]
 
 
+def _infonet_peer_url_allowed(peer_url: str) -> bool:
+    if not _infonet_private_transport_required():
+        return True
+    return _is_private_infonet_transport(peer_transport_kind(peer_url))
+
+
+def _filter_infonet_peer_urls(peer_urls: list[str]) -> list[str]:
+    if not _infonet_private_transport_required():
+        return peer_urls
+    return [peer_url for peer_url in peer_urls if _infonet_peer_url_allowed(peer_url)]
+
+
+def _infonet_peer_requests_proxies(normalized_peer_url: str) -> dict[str, str] | None:
+    """Return requests proxy settings for a sync/push peer, enforcing private policy."""
+    transport = peer_transport_kind(normalized_peer_url)
+    if _infonet_private_transport_required() and not _is_private_infonet_transport(transport):
+        raise RuntimeError(_infonet_private_transport_error())
+    if transport != "onion":
+        return None
+    if not bool(get_settings().MESH_ARTI_ENABLED):
+        raise RuntimeError("onion peer requests require Arti to be enabled")
+    from services.wormhole_supervisor import _check_arti_ready
+
+    if not _check_arti_ready():
+        raise RuntimeError("onion peer requests require a ready Arti transport")
+    socks_port = int(get_settings().MESH_ARTI_SOCKS_PORT or 9050)
+    proxy = f"socks5h://127.0.0.1:{socks_port}"
+    return {"http": proxy, "https": proxy}
+
+
+def _local_infonet_peer_url() -> str:
+    """Return this node's advertised peer URL for HMAC peer authentication."""
+    configured = normalize_peer_url(str(getattr(get_settings(), "MESH_PUBLIC_PEER_URL", "") or ""))
+    if configured:
+        return configured
+    try:
+        from services.tor_hidden_service import tor_service
+
+        return normalize_peer_url(str(tor_service.onion_address or ""))
+    except Exception:
+        return ""
+
+
 def _ensure_infonet_private_transport_ready(reason: str = "") -> bool:
     """Warm the local onion transport before private Infonet sync.
 
@@ -1257,6 +1307,13 @@ def _refresh_node_peer_store(*, now: float | None = None) -> dict[str, Any]:
     operator_peers = configured_relay_peer_urls()
     bootstrap_seed_peers = _configured_bootstrap_seed_peer_urls()
     skipped_clearnet_peers = 0
+    pruned_clearnet_peers = 0
+    if private_transport_required:
+        for key, record in list(store._records.items()):
+            if _is_private_infonet_transport(str(getattr(record, "transport", "") or "")):
+                continue
+            del store._records[key]
+            pruned_clearnet_peers += 1
     for peer_url in operator_peers:
         transport = peer_transport_kind(peer_url)
         if not transport:
@@ -1364,6 +1421,7 @@ def _refresh_node_peer_store(*, now: float | None = None) -> dict[str, Any]:
         "node_mode": mode,
         "private_transport_required": private_transport_required,
         "skipped_clearnet_peer_count": skipped_clearnet_peers,
+        "pruned_clearnet_peer_count": pruned_clearnet_peers,
         "manifest_loaded": manifest is not None,
         "manifest_signer_id": manifest.signer_id if manifest is not None else "",
         "manifest_valid_until": int(manifest.valid_until or 0) if manifest is not None else 0,
@@ -1384,6 +1442,28 @@ def _materialize_local_infonet_state() -> None:
     from services.mesh.mesh_hashchain import infonet
 
     infonet.ensure_materialized()
+    try:
+        _hydrate_gate_store_from_chain(list(infonet.events))
+        _hydrate_dm_relay_from_chain(list(infonet.events))
+    except Exception:
+        pass
+
+
+class PeerSyncHTTPError(RuntimeError):
+    def __init__(self, status_code: int, detail: str, *, retry_after_s: int = 0):
+        self.status_code = int(status_code or 0)
+        self.retry_after_s = int(retry_after_s or 0)
+        message = str(detail or f"HTTP {self.status_code}").strip()
+        if not message.upper().startswith("HTTP"):
+            message = f"HTTP {self.status_code}: {message}"
+        super().__init__(message)
+
+
+def _parse_retry_after_seconds(value: str) -> int:
+    try:
+        return max(0, int(float(str(value or "").strip())))
+    except Exception:
+        return 0
 
 
 def _peer_sync_response(peer_url: str, body: dict[str, Any]) -> dict[str, Any]:
@@ -1446,7 +1526,8 @@ def _peer_sync_response(peer_url: str, body: dict[str, Any]) -> dict[str, Any]:
         raise ValueError(f"peer sync returned non-JSON response ({response.status_code})") from exc
     if response.status_code != 200:
         detail = str(payload.get("detail", "") or f"HTTP {response.status_code}").strip()
-        raise ValueError(detail or f"HTTP {response.status_code}")
+        retry_after_s = _parse_retry_after_seconds(response.headers.get("Retry-After", ""))
+        raise PeerSyncHTTPError(response.status_code, detail, retry_after_s=retry_after_s)
     if not isinstance(payload, dict):
         raise ValueError("peer sync returned malformed payload")
     return payload
@@ -1485,6 +1566,46 @@ def _hydrate_gate_store_from_chain(events: list[dict]) -> int:
     return count
 
 
+def _hydrate_dm_relay_from_chain(events: list[dict]) -> int:
+    """Copy accepted dm_message chain events into the local encrypted DM relay."""
+    import hashlib
+
+    from services.mesh.mesh_dm_relay import dm_relay
+    from services.mesh.mesh_hashchain import infonet
+
+    count = 0
+    for evt in events:
+        if evt.get("event_type") != "dm_message":
+            continue
+        event_id = str(evt.get("event_id", "") or "").strip()
+        if not event_id or event_id not in infonet.event_index:
+            continue
+        canonical = infonet.events[infonet.event_index[event_id]]
+        payload = canonical.get("payload") if isinstance(canonical.get("payload"), dict) else {}
+        sender_token_hash = hashlib.sha256(
+            f"hashchain-dm-sender|{event_id}|{canonical.get('node_id', '')}".encode("utf-8")
+        ).hexdigest()
+        try:
+            result = dm_relay.deposit(
+                sender_id=str(canonical.get("node_id", "") or ""),
+                raw_sender_id=str(canonical.get("node_id", "") or ""),
+                recipient_id=str(payload.get("recipient_id", "") or ""),
+                ciphertext=str(payload.get("ciphertext", "") or ""),
+                msg_id=str(payload.get("msg_id", "") or ""),
+                delivery_class=str(payload.get("delivery_class", "") or ""),
+                recipient_token=str(payload.get("recipient_token", "") or "") or None,
+                sender_seal=str(payload.get("sender_seal", "") or ""),
+                sender_token_hash=sender_token_hash,
+                payload_format=str(payload.get("format", "dm1") or "dm1"),
+                session_welcome=str(payload.get("session_welcome", "") or ""),
+            )
+            if result.get("ok"):
+                count += 1
+        except Exception:
+            pass
+    return count
+
+
 def _sync_from_peer(
     peer_url: str,
     *,
@@ -1538,6 +1659,7 @@ def _sync_from_peer(
             return True, "", False, 0
         result = infonet.ingest_events(events)
         _hydrate_gate_store_from_chain(events)
+        _hydrate_dm_relay_from_chain(events)
         rejected = list(result.get("rejected", []) or [])
         if rejected:
             return False, f"sync ingest rejected {len(rejected)} event(s)", False, 0
@@ -1600,6 +1722,8 @@ def _run_public_sync_cycle() -> SyncWorkerState:
 
     last_error = "sync failed"
     for record in peers:
+        retry_after_s = 0
+        http_status_code = 0
         started = begin_sync(
             current_state,
             peer_url=record.peer_url,
@@ -1610,6 +1734,17 @@ def _run_public_sync_cycle() -> SyncWorkerState:
             set_sync_state(started)
         try:
             ok, error, forked, retry_after_s = _sync_from_peer(record.peer_url)
+        except PeerSyncHTTPError as exc:
+            # _sync_from_peer catches PeerSyncRateLimited internally (4-tuple
+            # path for 429 with Retry-After). Other non-200 statuses surface
+            # here as PeerSyncHTTPError — pull retry_after_s + status off it
+            # so the cooldown calculation below can honor server hints even
+            # for non-429 throttling responses.
+            ok = False
+            error = str(exc)
+            forked = False
+            retry_after_s = int(exc.retry_after_s or 0)
+            http_status_code = int(exc.status_code or 0)
         except Exception as exc:
             ok = False
             error = str(exc or type(exc).__name__)
@@ -1640,6 +1775,10 @@ def _run_public_sync_cycle() -> SyncWorkerState:
                 getattr(settings, "MESH_BOOTSTRAP_SEED_FAILURE_COOLDOWN_S", cooldown_s)
                 or cooldown_s
             )
+        if http_status_code == 429:
+            failure_count = max(int(getattr(record, "failure_count", 0) or 0), current_state.consecutive_failures)
+            exponential_429_s = min(900, 60 * (2 ** min(failure_count, 4)))
+            cooldown_s = max(cooldown_s, retry_after_s, exponential_429_s)
         store.mark_failure(
             record.peer_url,
             "sync",
@@ -1650,7 +1789,7 @@ def _run_public_sync_cycle() -> SyncWorkerState:
         store.save()
         failure_backoff_s = int(settings.MESH_SYNC_FAILURE_BACKOFF_S or 60)
         if is_seed_peer:
-            failure_backoff_s = min(failure_backoff_s, max(1, cooldown_s))
+            failure_backoff_s = max(failure_backoff_s, max(1, cooldown_s))
         updated = finish_sync(
             started,
             ok=False,
@@ -1750,7 +1889,7 @@ def _propagate_public_event_to_peers(event_dict: dict[str, Any]) -> None:
 
     if not _participant_node_enabled():
         return
-    if not authenticated_push_peer_urls():
+    if not _filter_infonet_peer_urls(authenticated_push_peer_urls()):
         return
 
     envelope = MeshEnvelope(
@@ -1784,6 +1923,45 @@ def _schedule_public_event_propagation(event_dict: dict[str, Any]) -> None:
     ).start()
 
 
+def _infonet_node_runtime_requested() -> bool:
+    return (not _MESH_ONLY) or _HEADLESS_MESH_NODE_RUNTIME
+
+
+def _start_infonet_node_runtime(reason: str = "startup") -> None:
+    """Start sync/push/pull workers for participant nodes."""
+    global _NODE_PUBLIC_EVENT_HOOK_REGISTERED, _NODE_RUNTIME_THREADS_STARTED
+
+    if not _infonet_node_runtime_requested():
+        return
+    try:
+        from services.mesh.mesh_hashchain import register_public_event_append_hook
+
+        _materialize_local_infonet_state()
+        _refresh_node_peer_store()
+        if _node_runtime_supported():
+            if not _participant_node_enabled():
+                logger.info("Infonet participant auto-enabled for private seed sync")
+                _set_participant_node_enabled(True)
+            threading.Thread(
+                target=lambda: _ensure_infonet_private_transport_ready(reason),
+                daemon=True,
+                name="infonet-private-transport-warmup",
+            ).start()
+            _NODE_SYNC_STOP.clear()
+            if not _NODE_RUNTIME_THREADS_STARTED:
+                threading.Thread(target=_public_infonet_sync_loop, daemon=True).start()
+                threading.Thread(target=_http_peer_push_loop, daemon=True).start()
+                threading.Thread(target=_http_gate_push_loop, daemon=True).start()
+                threading.Thread(target=_http_gate_pull_loop, daemon=True).start()
+                _NODE_RUNTIME_THREADS_STARTED = True
+            _kick_public_sync_background(reason)
+        if not _NODE_PUBLIC_EVENT_HOOK_REGISTERED:
+            register_public_event_append_hook(_schedule_public_event_propagation)
+            _NODE_PUBLIC_EVENT_HOOK_REGISTERED = True
+    except Exception as e:
+        logger.warning(f"Node bootstrap runtime failed to initialize: {e}")
+
+
 # ─── Background HTTP Peer Push Worker ────────────────────────────────────
 # Runs alongside the sync loop.  Every PUSH_INTERVAL seconds, batches new
 # Infonet events and sends them via HMAC-authenticated POST to push peers.
@@ -1791,6 +1969,7 @@ def _schedule_public_event_propagation(event_dict: dict[str, Any]) -> None:
 _PEER_PUSH_INTERVAL_S = 10
 _PEER_PUSH_BATCH_SIZE = 50
 _peer_push_last_index: dict[str, int] = {}  # peer_url → last pushed event index
+_INFONET_SYNC_RATE_LIMIT = "600/minute"
 
 
 def _http_peer_push_loop() -> None:
@@ -1812,7 +1991,7 @@ def _http_peer_push_loop() -> None:
             # loop on the global secret being set — an install that only
             # configures per-peer secrets is now valid.
 
-            peers = authenticated_push_peer_urls()
+            peers = _filter_infonet_peer_urls(authenticated_push_peer_urls())
             if not peers:
                 _NODE_SYNC_STOP.wait(_PEER_PUSH_INTERVAL_S)
                 continue
@@ -1840,7 +2019,8 @@ def _http_peer_push_loop() -> None:
                         ensure_ascii=False,
                     ).encode("utf-8")
 
-                    peer_key = resolve_peer_key_for_url(normalized)
+                    sender_url = _local_infonet_peer_url()
+                    peer_key = resolve_peer_key_for_url(sender_url)
                     if not peer_key:
                         continue
                     import hmac as _hmac_mod2
@@ -1848,14 +2028,21 @@ def _http_peer_push_loop() -> None:
                     hmac_hex = _hmac_mod2.new(peer_key, body_bytes, _hashlib_mod2.sha256).hexdigest()
 
                     timeout = int(get_settings().MESH_RELAY_PUSH_TIMEOUT_S or 10)
-                    resp = _requests.post(
-                        f"{normalized}/api/mesh/infonet/peer-push",
-                        data=body_bytes,
-                        headers={
+                    proxies = _infonet_peer_requests_proxies(normalized)
+                    request_kwargs: dict[str, Any] = {
+                        "data": body_bytes,
+                        "headers": {
                             "Content-Type": "application/json",
+                            "X-Peer-Url": sender_url,
                             "X-Peer-HMAC": hmac_hex,
                         },
-                        timeout=timeout,
+                        "timeout": timeout,
+                    }
+                    if proxies:
+                        request_kwargs["proxies"] = proxies
+                    resp = _requests.post(
+                        f"{normalized}/api/mesh/infonet/peer-push",
+                        **request_kwargs,
                     )
                     if resp.status_code == 200:
                         _peer_push_last_index[normalized] = last_idx + len(batch)
@@ -1895,7 +2082,7 @@ def _http_gate_pull_loop() -> None:
 
             # Issue #256: per-peer key resolution; see _http_peer_push_loop.
 
-            peers = authenticated_push_peer_urls()
+            peers = _filter_infonet_peer_urls(authenticated_push_peer_urls())
             if not peers:
                 _NODE_SYNC_STOP.wait(_GATE_PULL_INTERVAL_S)
                 continue
@@ -1905,7 +2092,8 @@ def _http_gate_pull_loop() -> None:
                 if not normalized:
                     continue
 
-                peer_key = resolve_peer_key_for_url(normalized)
+                sender_url = _local_infonet_peer_url()
+                peer_key = resolve_peer_key_for_url(sender_url)
                 if not peer_key:
                     continue
 
@@ -1925,14 +2113,21 @@ def _http_gate_pull_loop() -> None:
                     discovery_hmac = _hmac_pull.new(peer_key, discovery_body, _hashlib_pull.sha256).hexdigest()
 
                     timeout = int(get_settings().MESH_RELAY_PUSH_TIMEOUT_S or 10)
-                    resp = _requests.post(
-                        f"{normalized}/api/mesh/gate/peer-pull",
-                        data=discovery_body,
-                        headers={
+                    proxies = _infonet_peer_requests_proxies(normalized)
+                    discovery_kwargs: dict[str, Any] = {
+                        "data": discovery_body,
+                        "headers": {
                             "Content-Type": "application/json",
+                            "X-Peer-Url": sender_url,
                             "X-Peer-HMAC": discovery_hmac,
                         },
-                        timeout=timeout,
+                        "timeout": timeout,
+                    }
+                    if proxies:
+                        discovery_kwargs["proxies"] = proxies
+                    resp = _requests.post(
+                        f"{normalized}/api/mesh/gate/peer-pull",
+                        **discovery_kwargs,
                     )
                     if resp.status_code != 200:
                         continue
@@ -1962,14 +2157,20 @@ def _http_gate_pull_loop() -> None:
 
                         pull_hmac = _hmac_pull.new(peer_key, pull_body, _hashlib_pull.sha256).hexdigest()
 
-                        pull_resp = _requests.post(
-                            f"{normalized}/api/mesh/gate/peer-pull",
-                            data=pull_body,
-                            headers={
+                        pull_kwargs: dict[str, Any] = {
+                            "data": pull_body,
+                            "headers": {
                                 "Content-Type": "application/json",
+                                "X-Peer-Url": sender_url,
                                 "X-Peer-HMAC": pull_hmac,
                             },
-                            timeout=timeout,
+                            "timeout": timeout,
+                        }
+                        if proxies:
+                            pull_kwargs["proxies"] = proxies
+                        pull_resp = _requests.post(
+                            f"{normalized}/api/mesh/gate/peer-pull",
+                            **pull_kwargs,
                         )
                         if pull_resp.status_code != 200:
                             continue
@@ -2020,7 +2221,7 @@ def _http_gate_push_loop() -> None:
 
             # Issue #256: per-peer key resolution; see _http_peer_push_loop.
 
-            peers = authenticated_push_peer_urls()
+            peers = _filter_infonet_peer_urls(authenticated_push_peer_urls())
             if not peers:
                 _NODE_SYNC_STOP.wait(_PEER_PUSH_INTERVAL_S)
                 continue
@@ -2033,7 +2234,8 @@ def _http_gate_push_loop() -> None:
                 if not normalized:
                     continue
 
-                peer_key = resolve_peer_key_for_url(normalized)
+                sender_url = _local_infonet_peer_url()
+                peer_key = resolve_peer_key_for_url(sender_url)
                 if not peer_key:
                     continue
 
@@ -2064,14 +2266,21 @@ def _http_gate_push_loop() -> None:
                         hmac_hex = _hmac_mod3.new(peer_key, body_bytes, _hashlib_mod3.sha256).hexdigest()
 
                         timeout = int(get_settings().MESH_RELAY_PUSH_TIMEOUT_S or 10)
-                        resp = _requests.post(
-                            f"{normalized}/api/mesh/gate/peer-push",
-                            data=body_bytes,
-                            headers={
+                        proxies = _infonet_peer_requests_proxies(normalized)
+                        request_kwargs: dict[str, Any] = {
+                            "data": body_bytes,
+                            "headers": {
                                 "Content-Type": "application/json",
+                                "X-Peer-Url": sender_url,
                                 "X-Peer-HMAC": hmac_hex,
                             },
-                            timeout=timeout,
+                            "timeout": timeout,
+                        }
+                        if proxies:
+                            request_kwargs["proxies"] = proxies
+                        resp = _requests.post(
+                            f"{normalized}/api/mesh/gate/peer-push",
+                            **request_kwargs,
                         )
                         if resp.status_code == 200:
                             peer_counts[gate_id] = last + len(batch)
@@ -2413,32 +2622,8 @@ async def lifespan(app: FastAPI):
             daemon=True,
             name="wormhole-startup-sync",
         ).start()
-        try:
-            from services.mesh.mesh_hashchain import register_public_event_append_hook
 
-            _materialize_local_infonet_state()
-            _refresh_node_peer_store()
-            if _node_runtime_supported():
-                if not _participant_node_enabled():
-                    logger.info("Infonet participant auto-enabled for private seed sync")
-                    _set_participant_node_enabled(True)
-                threading.Thread(
-                    target=lambda: _ensure_infonet_private_transport_ready("startup"),
-                    daemon=True,
-                    name="infonet-private-transport-warmup",
-                ).start()
-                _NODE_SYNC_STOP.clear()
-                threading.Thread(target=_public_infonet_sync_loop, daemon=True).start()
-                _kick_public_sync_background("startup")
-                threading.Thread(target=_http_peer_push_loop, daemon=True).start()
-                threading.Thread(target=_http_gate_push_loop, daemon=True).start()
-                threading.Thread(target=_http_gate_pull_loop, daemon=True).start()
-            global _NODE_PUBLIC_EVENT_HOOK_REGISTERED
-            if not _NODE_PUBLIC_EVENT_HOOK_REGISTERED:
-                register_public_event_append_hook(_schedule_public_event_propagation)
-                _NODE_PUBLIC_EVENT_HOOK_REGISTERED = True
-        except Exception as e:
-            logger.warning(f"Node bootstrap runtime failed to initialize: {e}")
+    _start_infonet_node_runtime("startup")
 
     if not _MESH_ONLY:
         # Prime the static route/airport database from vrs-standing-data.adsb.lol
@@ -2677,6 +2862,91 @@ def _redact_public_event(event: dict) -> dict:
     return _redact_vote_gate(_redact_key_rotate_payload(_redact_gate_metadata(event)))
 
 
+def _is_loopback_host(host: str) -> bool:
+    value = str(host or "").strip().lower()
+    if not value:
+        return False
+    if value.startswith("[") and "]" in value:
+        value = value[1 : value.index("]")]
+    if ":" in value and value.count(":") == 1:
+        value = value.rsplit(":", 1)[0]
+    if value in {"localhost", "ip6-localhost"}:
+        return True
+    try:
+        return ipaddress.ip_address(value).is_loopback
+    except ValueError:
+        return False
+
+
+def _is_onion_host(host: str) -> bool:
+    value = str(host or "").strip().lower()
+    if not value:
+        return False
+    if ":" in value and value.count(":") == 1:
+        value = value.rsplit(":", 1)[0]
+    return value.endswith(".onion")
+
+
+def _forwarded_for_hosts(request) -> list[str]:
+    headers = getattr(request, "headers", {}) or {}
+    hosts: list[str] = []
+    x_forwarded_for = str(headers.get("x-forwarded-for", "") or "")
+    hosts.extend(part.strip() for part in x_forwarded_for.split(",") if part.strip())
+    forwarded = str(headers.get("forwarded", "") or "")
+    for section in forwarded.split(","):
+        for item in section.split(";"):
+            key, sep, value = item.strip().partition("=")
+            if sep and key.strip().lower() == "for":
+                hosts.append(value.strip().strip('"').strip("[]"))
+    return hosts
+
+
+def _request_appears_private_infonet_transport(request) -> bool:
+    """Return whether a sync request is safe to carry private ledger events.
+
+    This is intentionally fail-closed for the private event surface only. A
+    questionable request still gets public events; gate/DM ciphertext simply
+    stays out of the response.
+    """
+    if not _infonet_private_transport_required() or request is None:
+        return False
+
+    forwarded_hosts = _forwarded_for_hosts(request)
+    if forwarded_hosts and any(not (_is_loopback_host(host) or _is_onion_host(host)) for host in forwarded_hosts):
+        return False
+
+    client = getattr(request, "client", None)
+    client_host = str(getattr(client, "host", "") or "")
+    headers = getattr(request, "headers", {}) or {}
+    host_header = str(headers.get("host", "") or "")
+    url_host = str(getattr(getattr(request, "url", None), "hostname", "") or "")
+    return any(
+        (
+            _is_loopback_host(client_host),
+            _is_loopback_host(host_header),
+            _is_loopback_host(url_host),
+            _is_onion_host(host_header),
+            _is_onion_host(url_host),
+        )
+    )
+
+
+def _infonet_sync_response_events(events: list[dict], request=None) -> list[dict]:
+    """Build the sync event surface for the current transport policy."""
+    include_private = _request_appears_private_infonet_transport(request)
+    response: list[dict] = []
+    for event in events:
+        if not isinstance(event, dict):
+            continue
+        event_type = str(event.get("event_type", "") or "")
+        if event_type in {"gate_message", "dm_message"}:
+            if include_private:
+                response.append(dict(event))
+            continue
+        response.append(_redact_public_event(event))
+    return response
+
+
 def _trusted_gate_reply_to(event: dict) -> str:
     if not isinstance(event, dict):
         return ""
@@ -5261,32 +5531,15 @@ def _submit_gate_message_envelope(request: Request, gate_id: str, body: dict[str
     if not cooldown_ok:
         return {"ok": False, "detail": cooldown_reason}
 
-    # Advance sequence counter (replay protection) without appending to
-    # the public infonet chain — gate messages are private.
-    try:
-        from services.mesh.mesh_hashchain import infonet, gate_store
-
-        seq_ok, seq_reason = _validate_private_signed_sequence(
-            infonet,
-            sender_id,
-            sequence,
-            domain="gate_message",
-        )
-        if not seq_ok:
-            return {"ok": False, "detail": seq_reason}
-    except ValueError as exc:
-        return {"ok": False, "detail": str(exc)}
-    except Exception:
-        logger.exception("Failed to advance sequence for gate message")
-        return {"ok": False, "detail": "Failed to record gate message"}
-
     gate_manager.record_message(gate_id)
     _record_gate_post_cooldown(sender_id, gate_id)
     logger.info("Encrypted gate message accepted on obfuscated gate plane")
 
-    # Build gate event and store in gate_store (private — not on public chain).
+    # Build and commit the encrypted gate event to the private Infonet ledger.
+    # The main hashchain is the durable propagation surface; gate_store is the
+    # local materialized view used by the existing decrypt/UI path.
     try:
-        from services.mesh.mesh_hashchain import _private_gate_event_id
+        from services.mesh.mesh_hashchain import infonet
         import time as _time
 
         store_payload = dict(gate_payload)
@@ -5308,19 +5561,24 @@ def _submit_gate_message_envelope(request: Request, gate_id: str, body: dict[str
             "public_key_algo": public_key_algo,
             "protocol_version": protocol_version or PROTOCOL_VERSION,
         }
-        gate_event["event_id"] = _private_gate_event_id(gate_id, sender_id, sequence, gate_event)
+        gate_event = infonet.append_private_gate_message(
+            node_id=sender_id,
+            payload=store_payload,
+            signature=signature,
+            sequence=sequence,
+            public_key=public_key,
+            public_key_algo=public_key_algo,
+            protocol_version=protocol_version or PROTOCOL_VERSION,
+            timestamp=float(gate_event.get("timestamp", 0) or 0),
+        )
+    except ValueError as exc:
+        return {"ok": False, "detail": str(exc)}
     except Exception:
-        logger.exception("Failed to prepare private gate message for queued release")
+        logger.exception("Failed to append gate message to private Infonet ledger")
         return {"ok": False, "detail": "Failed to record gate message"}
 
-    # Append to the local gate_store immediately. The gate_store is a
-    # per-node persistent ciphertext chain; writing to it is a local
-    # operation with no network dependency. Previously this happened only
-    # inside the release worker's attempt_private_release path, which
-    # meant messages sat in the outbox — invisible to the author and the
-    # gate UI — until the transport tier reached the release floor.
-    # Decoupling local visibility from network fan-out: append locally now,
-    # queue the release for network propagation when the lane is ready.
+    # Append to the local gate_store immediately so the author sees the same
+    # materialized gate view that peers will hydrate after private sync.
     try:
         from services.mesh.mesh_hashchain import gate_store
 
@@ -5447,7 +5705,7 @@ async def infonet_locator(request: Request, limit: int = Query(32, ge=4, le=128)
 
 
 @app.post("/api/mesh/infonet/sync")
-@limiter.limit("30/minute")
+@limiter.limit(_INFONET_SYNC_RATE_LIMIT)
 @mesh_write_exempt(MeshWriteExemption.PEER_GOSSIP)
 async def infonet_sync_post(
     request: Request,
@@ -5500,8 +5758,7 @@ async def infonet_sync_post(
     elif matched_hash == GENESIS_HASH and len(locator) > 1:
         forked = True
 
-    # Filter out legacy gate_message events — not part of the public sync surface.
-    events = [_redact_public_event(e) for e in events if e.get("event_type") != "gate_message"]
+    events = _infonet_sync_response_events(events, request=request)
 
     response = {
         "events": events,
@@ -5564,7 +5821,7 @@ async def mesh_rns_status(request: Request):
 
 
 @app.get("/api/mesh/infonet/sync")
-@limiter.limit("30/minute")
+@limiter.limit(_INFONET_SYNC_RATE_LIMIT)
 async def infonet_sync(
     request: Request,
     after_hash: str = "",
@@ -5602,8 +5859,7 @@ async def infonet_sync(
         )
     base = after_hash or GENESIS_HASH
     events = infonet.get_events_after(base, limit=limit)
-    # Filter out legacy gate_message events — not part of the public sync surface.
-    events = [_redact_public_event(e) for e in events if e.get("event_type") != "gate_message"]
+    events = _infonet_sync_response_events(events, request=request)
     return {
         "events": events,
         "after_hash": base,
@@ -5642,6 +5898,7 @@ async def infonet_ingest(request: Request):
 
     result = infonet.ingest_events(events)
     _hydrate_gate_store_from_chain(events)
+    _hydrate_dm_relay_from_chain(events)
     return {"ok": True, **result}
 
 
@@ -5682,6 +5939,7 @@ async def infonet_peer_push(request: Request):
 
     result = infonet.ingest_events(events)
     _hydrate_gate_store_from_chain(events)
+    _hydrate_dm_relay_from_chain(events)
     return {"ok": True, **result}
 
 
@@ -6241,6 +6499,12 @@ async def infonet_event(request: Request, event_id: str):
                 )
             return _strip_gate_for_access(evt, access)
         return {"ok": False, "detail": "Event not found"}
+    if evt.get("event_type") == "dm_message":
+        return await _private_plane_refusal_response(
+            request,
+            status_code=403,
+            payload=_private_plane_access_denied_payload(),
+        )
     if evt.get("event_type") == "gate_message":
         gate_id = str(evt.get("payload", {}).get("gate", "") or evt.get("gate", "") or "").strip()
         access = _verify_gate_access(request, gate_id) if gate_id else ""
@@ -6265,7 +6529,7 @@ async def infonet_node_events(
     from services.mesh.mesh_hashchain import infonet
 
     events = infonet.get_events_by_node(node_id, limit=limit)
-    events = [e for e in events if e.get("event_type") != "gate_message"]
+    events = [e for e in events if e.get("event_type") not in {"gate_message", "dm_message"}]
     events = [_redact_public_event(e) for e in infonet.decorate_events(events)]
     events = _redact_public_node_history(
         events,
@@ -6290,7 +6554,7 @@ async def infonet_events_by_type(
     else:
         events = list(reversed(infonet.events))
         events = events[offset : offset + limit]
-    events = [e for e in events if e.get("event_type") != "gate_message"]
+    events = [e for e in events if e.get("event_type") not in {"gate_message", "dm_message"}]
     events = [_redact_public_event(e) for e in infonet.decorate_events(events)]
     return {
         "events": events,
@@ -7028,6 +7292,7 @@ async def _dm_send_from_signed_request(request: Request):
     relay_salt_hex = str(body.get("relay_salt", "") or "").strip().lower()
     msg_id = str(body.get("msg_id", "")).strip()
     timestamp = _safe_int(body.get("timestamp", 0) or 0)
+    sequence = _safe_int(body.get("sequence", 0) or 0)
     nonce = str(body.get("nonce", "")).strip()
 
     if not sender_id or not recipient_id or not ciphertext or not msg_id or not timestamp:
@@ -7101,7 +7366,7 @@ async def _dm_send_from_signed_request(request: Request):
         ok_seq, seq_reason = _validate_private_signed_sequence(
             infonet,
             sender_id,
-            int(body.get("sequence", 0) or 0),
+            sequence,
             domain="dm_send",
         )
         if not ok_seq:
@@ -7135,7 +7400,47 @@ async def _dm_send_from_signed_request(request: Request):
         "sender_seal": sender_seal,
         "relay_salt": relay_salt_hex,
     }
+    hashchain_spool: dict[str, Any] = {"ok": False, "detail": "not attempted"}
+    try:
+        from services.mesh.mesh_hashchain import infonet
+
+        chain_payload = dict(prepared.payload if prepared is not None else {})
+        if not chain_payload:
+            chain_payload = {
+                "recipient_id": recipient_id,
+                "delivery_class": delivery_class,
+                "recipient_token": recipient_token if delivery_class == "shared" else "",
+                "ciphertext": ciphertext,
+                "msg_id": msg_id,
+                "timestamp": timestamp,
+                "format": payload_format,
+            }
+        chain_payload["transport_lock"] = "private_strong"
+        chain_event = infonet.append_private_dm_message(
+            node_id=sender_id,
+            payload=chain_payload,
+            signature=str(prepared.signature if prepared is not None else body.get("signature", "") or ""),
+            sequence=sequence,
+            public_key=str(prepared.public_key if prepared is not None else body.get("public_key", "") or ""),
+            public_key_algo=str(
+                prepared.public_key_algo if prepared is not None else body.get("public_key_algo", "") or ""
+            ),
+            protocol_version=str(
+                prepared.protocol_version if prepared is not None else body.get("protocol_version", "") or ""
+            )
+            or PROTOCOL_VERSION,
+            timestamp=float(timestamp or time.time()),
+        )
+        _hydrate_dm_relay_from_chain([chain_event])
+        hashchain_spool = {
+            "ok": True,
+            "event_id": str(chain_event.get("event_id", "") or ""),
+            "limit": 2,
+        }
+    except Exception as exc:
+        hashchain_spool = {"ok": False, "detail": str(exc) or type(exc).__name__}
     queued_result = _queue_dm_release(current_tier=tier, payload=release_payload)
+    queued_result["hashchain_spool"] = hashchain_spool
     if transport_upgrade_pending:
         queued_result["private_transport_pending"] = True
     return queued_result
@@ -9111,6 +9416,11 @@ async def api_get_node_settings(request: Request):
 async def api_set_node_settings(request: Request, body: NodeSettingsUpdate):
     _refresh_node_peer_store()
     if bool(body.enabled):
+        if _infonet_private_transport_required() and not _ensure_infonet_private_transport_ready("operator_enable"):
+            return JSONResponse(
+                {"ok": False, "detail": _infonet_private_transport_error()},
+                status_code=503,
+            )
         try:
             from services.transport_lane_isolation import disable_public_mesh_lane
 
@@ -9119,6 +9429,7 @@ async def api_set_node_settings(request: Request, body: NodeSettingsUpdate):
             logger.warning("Failed to disable public Mesh while enabling private node: %s", exc)
     result = _set_participant_node_enabled(bool(body.enabled))
     if bool(body.enabled):
+        _start_infonet_node_runtime("operator_enable")
         _kick_public_sync_background("operator_enable")
     return result
 

backend/pyproject.toml

@@ -7,7 +7,7 @@ py-modules = []
 
 [project]
 name = "backend"
-version = "0.9.79"
+version = "0.9.81"
 requires-python = ">=3.10"
 dependencies = [
     "apscheduler==3.10.3",
@@ -43,7 +43,7 @@ dev = ["pytest>=8.3.4", "pytest-asyncio==0.25.0", "ruff>=0.9.0", "black>=24.0.0"
 
 [tool.ruff.lint]
 # The current backend carries historical style debt in large legacy modules.
-# Keep CI focused on actionable correctness checks for the v0.9.79 release.
+# Keep CI focused on actionable correctness checks for the v0.9.81 release.
 ignore = ["E401", "E402", "E701", "E731", "E741", "F401", "F402", "F541", "F811", "F841"]
 
 [tool.black]

backend/routers/ai_intel.py

@@ -1590,7 +1590,7 @@ async def agent_tool_manifest(request: Request):
 
     return {
         "ok": True,
-        "version": "0.9.79",
+        "version": "0.9.81",
         "access_tier": access_tier,
         "available_commands": available_commands,
         "transport": {
@@ -2226,7 +2226,7 @@ async def api_capabilities(request: Request):
     access_tier = str(get_settings().OPENCLAW_ACCESS_TIER or "restricted").strip().lower()
     return {
         "ok": True,
-        "version": "0.9.79",
+        "version": "0.9.81",
         "auth": {
             "method": "HMAC-SHA256",
             "headers": ["X-SB-Timestamp", "X-SB-Nonce", "X-SB-Signature"],

backend/routers/health.py

@@ -8,7 +8,7 @@ from services.data_fetcher import get_latest_data
 from services.schemas import HealthResponse
 import os
 
-APP_VERSION = os.environ.get("_HEALTH_APP_VERSION", "0.9.79")
+APP_VERSION = os.environ.get("_HEALTH_APP_VERSION", "0.9.81")
 
 router = APIRouter()
 
@@ -59,6 +59,12 @@ async def health_check(request: Request):
     # when the SPKI-pinned fallback is in effect. The data plane keeps
     # flowing (this is by design — see ais_proxy.js comments) but observers
     # who care about MITM-protection posture deserve a visible signal.
+    #
+    # Plus connectivity health (added 2026-05-23 when stream.aisstream.io
+    # went fully offline): ``connected`` tells the frontend whether ship
+    # data is actually flowing. When false, a banner explains that ships
+    # are unavailable due to an upstream outage — better than the user
+    # silently seeing an empty ocean and assuming we broke something.
     ais_status: dict = {}
     try:
         from services.ais_stream import ais_proxy_status
@@ -69,6 +75,15 @@ async def health_check(request: Request):
         # Don't override a worse top-level status if SLOs already failed,
         # but escalate ok -> degraded so the field surfaces in dashboards.
         top_status = "degraded"
+    # AIS_API_KEY not configured is "feature off", not "system broken" —
+    # so we only escalate when the operator opted into AIS (key set) AND
+    # the stream is currently offline.
+    if (
+        os.environ.get("AIS_API_KEY")
+        and ais_status.get("connected") is False
+        and top_status == "ok"
+    ):
+        top_status = "degraded"
 
     return {
         "status": top_status,

backend/routers/mesh_peer_sync.py

@@ -55,6 +55,12 @@ def _hydrate_gate_store_from_chain(events: list) -> int:
     return count
 
 
+def _hydrate_dm_relay_from_chain(events: list) -> int:
+    import main as _m
+
+    return int(_m._hydrate_dm_relay_from_chain(events))
+
+
 @router.post("/api/mesh/infonet/peer-push")
 @limiter.limit("30/minute")
 async def infonet_peer_push(request: Request):
@@ -82,6 +88,7 @@ async def infonet_peer_push(request: Request):
         return {"ok": True, "accepted": 0, "duplicates": 0, "rejected": []}
     result = infonet.ingest_events(events)
     _hydrate_gate_store_from_chain(events)
+    _hydrate_dm_relay_from_chain(events)
     return {"ok": True, **result}
 
 

backend/routers/mesh_public.py

@@ -65,6 +65,7 @@ from services.mesh.mesh_signed_events import (
 logger = logging.getLogger(__name__)
 
 router = APIRouter()
+_INFONET_SYNC_RATE_LIMIT = "600/minute"
 
 
 def _signed_body(request: Request) -> dict[str, Any]:
@@ -263,6 +264,19 @@ def _redact_public_event(event: dict) -> dict:
     return _redact_vote_gate(_redact_key_rotate_payload(_redact_gate_metadata(event)))
 
 
+def _infonet_private_transport_required() -> bool:
+    import main as _m
+
+    return bool(_m._infonet_private_transport_required())
+
+
+def _infonet_sync_response_events(events: list[dict], request=None) -> list[dict]:
+    """Build the sync event surface for the current transport policy."""
+    import main as _m
+
+    return _m._infonet_sync_response_events(events, request=request)
+
+
 def _trusted_gate_reply_to(event: dict) -> str:
     if not isinstance(event, dict):
         return ""
@@ -574,6 +588,12 @@ def _hydrate_gate_store_from_chain(events: list[dict]) -> int:
             pass
     return count
 
+
+def _hydrate_dm_relay_from_chain(events: list[dict]) -> int:
+    import main as _m
+
+    return int(_m._hydrate_dm_relay_from_chain(events))
+
 # --- Safe type helpers ---
 
 def _safe_int(val, default=0):
@@ -1531,7 +1551,7 @@ async def infonet_locator(request: Request, limit: int = Query(32, ge=4, le=128)
 
 
 @router.post("/api/mesh/infonet/sync")
-@limiter.limit("30/minute")
+@limiter.limit(_INFONET_SYNC_RATE_LIMIT)
 @mesh_write_exempt(MeshWriteExemption.PEER_GOSSIP)
 async def infonet_sync_post(
     request: Request,
@@ -1584,8 +1604,7 @@ async def infonet_sync_post(
     elif matched_hash == GENESIS_HASH and len(locator) > 1:
         forked = True
 
-    # Filter out legacy gate_message events — not part of the public sync surface.
-    events = [_redact_public_event(e) for e in events if e.get("event_type") != "gate_message"]
+    events = _infonet_sync_response_events(events, request=request)
 
     response = {
         "events": events,
@@ -1646,7 +1665,7 @@ async def mesh_rns_status(request: Request):
 
 
 @router.get("/api/mesh/infonet/sync")
-@limiter.limit("30/minute")
+@limiter.limit(_INFONET_SYNC_RATE_LIMIT)
 async def infonet_sync(
     request: Request,
     after_hash: str = "",
@@ -1684,8 +1703,7 @@ async def infonet_sync(
         )
     base = after_hash or GENESIS_HASH
     events = infonet.get_events_after(base, limit=limit)
-    # Filter out legacy gate_message events — not part of the public sync surface.
-    events = [_redact_public_event(e) for e in events if e.get("event_type") != "gate_message"]
+    events = _infonet_sync_response_events(events, request=request)
     return {
         "events": events,
         "after_hash": base,
@@ -1724,6 +1742,7 @@ async def infonet_ingest(request: Request):
 
     result = infonet.ingest_events(events)
     _hydrate_gate_store_from_chain(events)
+    _hydrate_dm_relay_from_chain(events)
     return {"ok": True, **result}
 
 
@@ -2279,6 +2298,12 @@ async def infonet_event(request: Request, event_id: str):
                 )
             return _strip_gate_for_access(evt, access)
         return {"ok": False, "detail": "Event not found"}
+    if evt.get("event_type") == "dm_message":
+        return await _private_plane_refusal_response(
+            request,
+            status_code=403,
+            payload=_private_plane_access_denied_payload(),
+        )
     if evt.get("event_type") == "gate_message":
         gate_id = str(evt.get("payload", {}).get("gate", "") or evt.get("gate", "") or "").strip()
         access = _verify_gate_access(request, gate_id) if gate_id else ""
@@ -2303,7 +2328,7 @@ async def infonet_node_events(
     from services.mesh.mesh_hashchain import infonet
 
     events = infonet.get_events_by_node(node_id, limit=limit)
-    events = [e for e in events if e.get("event_type") != "gate_message"]
+    events = [e for e in events if e.get("event_type") not in {"gate_message", "dm_message"}]
     events = [_redact_public_event(e) for e in infonet.decorate_events(events)]
     events = _redact_public_node_history(
         events,
@@ -2328,7 +2353,7 @@ async def infonet_events_by_type(
     else:
         events = list(reversed(infonet.events))
         events = events[offset : offset + limit]
-    events = [e for e in events if e.get("event_type") != "gate_message"]
+    events = [e for e in events if e.get("event_type") not in {"gate_message", "dm_message"}]
     events = [_redact_public_event(e) for e in infonet.decorate_events(events)]
     return {
         "events": events,

backend/services/ais_stream.py

@@ -350,19 +350,58 @@ _proxy_process = None
 # path during an upstream cert outage. Surfaced via ais_proxy_status() for
 # /api/health.
 _proxy_status: dict = {}
+# Upstream-connectivity telemetry (added when stream.aisstream.io went fully
+# offline on 2026-05-23). ``_last_msg_at`` is the unix timestamp of the most
+# recent vessel message received from the proxy. ``_proxy_spawn_count`` is
+# how many times we've started the node proxy; combined with no recent
+# messages it tells us the proxy is respawning in a tight loop because the
+# upstream is unreachable. Surfaced via ais_proxy_status() so the operator
+# can see "AIS is dead" instead of guessing whether it's their map filter,
+# their api key, or upstream.
+_last_msg_at: float = 0.0
+_proxy_spawn_count: int = 0
 _VESSEL_TRAIL_INTERVAL_S = 120
 _VESSEL_TRAIL_MAX_POINTS = 240
 
 
-def ais_proxy_status() -> dict:
-    """Return a copy of the latest ais_proxy.js status (issue #258).
+# How stale "last vessel message" can be before we consider the stream
+# disconnected. AISStream typically pushes multiple messages/sec, so a 60s
+# gap means something's wrong upstream or in transit.
+_AIS_CONNECTED_FRESHNESS_S = 60
 
-    Currently surfaces ``degraded_tls`` (bool) which is true when the
-    proxy is using SPKI-pinned fallback because AISStream's cert expired.
-    Returns an empty dict when no status has been received yet.
+
+def ais_proxy_status() -> dict:
+    """Return a copy of the latest ais_proxy.js status + connectivity health.
+
+    Fields:
+      * ``degraded_tls`` (bool, issue #258) — true when the proxy is using
+        SPKI-pinned fallback because AISStream's cert expired.
+      * ``connected`` (bool) — true when we received a vessel message in
+        the last ``_AIS_CONNECTED_FRESHNESS_S`` seconds.
+      * ``last_msg_age_seconds`` (int | None) — seconds since the last
+        vessel message; None if we've never received one.
+      * ``proxy_spawn_count`` (int) — how many times we've spawned the
+        node proxy. Sustained increases here without ``connected`` means
+        we're respawning in a tight loop because upstream is dead.
+
+    Returns an empty dict when called before the AIS subsystem starts
+    (e.g. during tests or when no API key is set).
     """
     with _vessels_lock:
-        return dict(_proxy_status)
+        status = dict(_proxy_status)
+        last = _last_msg_at
+        spawns = _proxy_spawn_count
+
+    now = time.time()
+    if last > 0:
+        last_age = int(now - last)
+        status["last_msg_age_seconds"] = last_age
+        status["connected"] = last_age <= _AIS_CONNECTED_FRESHNESS_S
+    else:
+        status["last_msg_age_seconds"] = None
+        status["connected"] = False
+    status["proxy_spawn_count"] = spawns
+    return status
 
 import os
 
@@ -588,8 +627,10 @@ def _ais_stream_loop():
                 env=proxy_env,
                 **popen_kwargs,
             )
+            global _proxy_spawn_count
             with _vessels_lock:
                 _proxy_process = process
+                _proxy_spawn_count += 1
 
             # Drain stderr in a background thread to prevent deadlock
             import threading
@@ -645,9 +686,15 @@ def _ais_stream_loop():
                 if not mmsi:
                     continue
 
+                # Telemetry: stamp the timestamp of the most recent real
+                # vessel message. ais_proxy_status() reads this to decide
+                # whether the stream is currently "connected" — i.e. has
+                # any data flowed in the last 60s.
+                global _last_msg_at
                 with _vessels_lock:
+                    _last_msg_at = time.time()
                     if mmsi not in _vessels:
-                        _vessels[mmsi] = {"_updated": time.time()}
+                        _vessels[mmsi] = {"_updated": _last_msg_at}
                     vessel = _vessels[mmsi]
 
                 # Update position from PositionReport or StandardClassBPositionReport

backend/services/config.py

@@ -32,6 +32,7 @@ class Settings(BaseSettings):
     MESH_ARTI_ENABLED: bool = False
     MESH_ARTI_SOCKS_PORT: int = 9050
     MESH_RELAY_PEERS: str = ""
+    MESH_PUBLIC_PEER_URL: str = ""
     # Bootstrap seeds are discovery hints, not authoritative network roots.
     # Nodes promote healthy discovered peers from the store/manifest over time.
     MESH_BOOTSTRAP_SEED_PEERS: str = "http://gqpbunqbgtkcqilvclm3xrkt3zowjyl3s62kkktvojgvxzizamvbrqid.onion:8000"

backend/services/constants.py

@@ -11,8 +11,13 @@ DEFAULT_TRAIL_TTL_S = 300  # 5 min - trail TTL for non-tracked flights
 HOLD_PATTERN_DEGREES = 300  # Total heading change to flag holding pattern
 GPS_JAMMING_NACP_THRESHOLD = 8  # NACp below this = degraded GPS signal
 GPS_JAMMING_GRID_SIZE = 1.0  # 1 degree grid for aggregation
-GPS_JAMMING_MIN_RATIO = 0.30  # 30% degraded aircraft to flag zone
-GPS_JAMMING_MIN_AIRCRAFT = 5  # Min aircraft in grid cell for statistical significance
+# Tuned 2026-05: previously 0.30 / 5 aircraft which — combined with the
+# -1 noise cushion in the detector AND the pre-fix nac_p==0 filter that
+# discarded jamming victims — meant the layer almost never lit up.
+# Lowering the bar so genuine jamming zones with sparser ADS-B coverage
+# clear (eastern Med, Russia/Ukraine border, Iran/Iraq).
+GPS_JAMMING_MIN_RATIO = 0.20  # 20% degraded aircraft to flag zone
+GPS_JAMMING_MIN_AIRCRAFT = 3  # Min aircraft in grid cell for statistical significance
 
 # ─── Network & Circuit Breaker ──────────────────────────────────────────────
 CIRCUIT_BREAKER_TTL_S = 120  # Skip domain for 2 min after total failure

backend/services/data_fetcher.py

@@ -777,6 +777,39 @@ def start_scheduler():
         misfire_grace_time=60,
     )
 
+    # Flight observation pruning — drops icao24 → first_seen_at entries we
+    # haven't seen in an hour. Same cadence as AIS prune for symmetry; the
+    # per-tick scan is O(in-flight aircraft) so it's cheap.
+    from services.fetchers.flight_observations import prune as _prune_flight_observations
+    _scheduler.add_job(
+        lambda: _run_task_with_health(_prune_flight_observations, "prune_flight_observations"),
+        "interval",
+        minutes=5,
+        id="flight_observation_prune",
+        max_instances=1,
+        misfire_grace_time=60,
+    )
+
+    # AISHub REST fallback — slow polling when the AISStream WebSocket
+    # primary is offline. Configurable interval via
+    # AISHUB_POLL_INTERVAL_MINUTES env (default 20 min). Operator must
+    # set AISHUB_USERNAME to opt in. The fetcher is gated internally on
+    # the primary being disconnected, so this job is cheap when the
+    # WebSocket is healthy (early-returns after a status check).
+    from services.fetchers.aishub_fallback import (
+        aishub_poll_interval_minutes,
+        fetch_aishub_vessels,
+    )
+    _aishub_interval = aishub_poll_interval_minutes()
+    _scheduler.add_job(
+        lambda: _run_task_with_health(fetch_aishub_vessels, "fetch_aishub_vessels"),
+        "interval",
+        minutes=_aishub_interval,
+        id="aishub_fallback",
+        max_instances=1,
+        misfire_grace_time=120,
+    )
+
     # Route database — bulk refresh from vrs-standing-data.adsb.lol every 5
     # days. Replaces the legacy /api/0/routeset POST (blocked under our UA,
     # and broken upstream). Airline schedules change on a quarterly cycle,
@@ -960,16 +993,19 @@ def start_scheduler():
         misfire_grace_time=600,
     )
 
-    # UAP sightings (NUFORC) — daily at 12:00 UTC
+    # UAP sightings (NUFORC) — weekly on Mondays at 12:00 UTC. The layer is a
+    # rolling last-60-days digest; refreshing once a week is enough cadence
+    # for human-readable map exploration and keeps load on nuforc.org light.
     _scheduler.add_job(
         lambda: _run_task_with_health(
             lambda: fetch_uap_sightings(force_refresh=True),
             "fetch_uap_sightings",
         ),
         "cron",
+        day_of_week="mon",
         hour=12,
         minute=0,
-        id="uap_sightings_daily",
+        id="uap_sightings_weekly",
         max_instances=1,
         misfire_grace_time=3600,
     )

backend/services/fetchers/aishub_fallback.py

@@ -0,0 +1,290 @@
+"""AISHub REST fallback for ship tracking when AISStream is unreachable.
+
+Background
+----------
+On 2026-05-23 ``stream.aisstream.io`` (the primary live AIS WebSocket feed)
+went fully offline. Backend's only ship signal vanished. This module polls
+``data.aishub.net``'s free REST API on a slow cadence (default 20 min) when
+the WebSocket primary is disconnected, so the ships layer doesn't go fully
+dark during upstream outages.
+
+Why 20 minutes
+--------------
+AISHub's free tier is rate-limited and explicitly asks consumers to be
+courteous. 20 minutes is well inside their limits, gives ships time to
+move enough to look "alive" on the map, and won't drain their service.
+Configurable via the ``AISHUB_POLL_INTERVAL_MINUTES`` env var (clamped to
+[1, 360]).
+
+Why slow vs primary
+-------------------
+This is degraded mode, not a replacement. A ship at 20 knots moves about
+6 nautical miles in 20 minutes — visible on the map but coarser than the
+real-time WebSocket signal. When AISStream comes back online, the
+WebSocket data will overwrite these records via the same ``_vessels``
+dict and ``source`` will flip from ``"aishub"`` back to upstream-live.
+
+Opt-in
+------
+Operator must set ``AISHUB_USERNAME`` (free registration at
+https://www.aishub.net/api). If unset, this fetcher is a no-op.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+import time
+from typing import Any
+
+from services.network_utils import fetch_with_curl
+
+logger = logging.getLogger(__name__)
+
+
+AISHUB_URL = "https://data.aishub.net/ws.php"
+
+
+def aishub_username() -> str:
+    return str(os.environ.get("AISHUB_USERNAME", "")).strip()
+
+
+def aishub_fallback_enabled() -> bool:
+    """Returns True only when the operator has registered with AISHub and
+    set ``AISHUB_USERNAME``. The presence of the username is the opt-in."""
+    return bool(aishub_username())
+
+
+def aishub_poll_interval_minutes() -> int:
+    """Default 20 minutes. Clamped to [1, 360] so a hostile or
+    misconfigured env var can't either hammer the upstream or silence the
+    fallback for a day."""
+    raw = os.environ.get("AISHUB_POLL_INTERVAL_MINUTES", "20")
+    try:
+        value = int(str(raw).strip())
+    except (TypeError, ValueError):
+        value = 20
+    return max(1, min(360, value))
+
+
+def _should_run_fallback() -> bool:
+    """Only run when the primary WebSocket is disconnected. Avoids stomping
+    over fresher live data when AISStream is healthy.
+
+    Returns False if:
+      * AISHub isn't configured (no username)
+      * AISStream primary is currently connected (recent vessel messages)
+
+    Returns True only when AIS is configured-but-down. The
+    ``proxy_spawn_count > 0`` guard means "the primary has at least tried
+    to run" — if the user set AISHUB_USERNAME but not AIS_API_KEY at all,
+    AISHub will still serve as a primary on its own slow cadence.
+    """
+    if not aishub_fallback_enabled():
+        return False
+    try:
+        from services.ais_stream import ais_proxy_status
+        status = ais_proxy_status() or {}
+    except Exception:
+        return True  # ais_stream not importable? still try AISHub.
+    # If the WebSocket primary is connected, skip the fallback — fresher
+    # data is already flowing.
+    if status.get("connected") is True:
+        return False
+    return True
+
+
+def _parse_aishub_response(payload: str) -> list[dict]:
+    """Parse the AISHub JSON response into a list of vessel records.
+
+    Successful response shape::
+
+        [
+            {"ERROR": false, "USERNAME": "...", "FORMAT": "1", "RECORDS": N},
+            [{"MMSI": ..., "LATITUDE": ..., "LONGITUDE": ..., ...}, ...]
+        ]
+
+    Error response shape::
+
+        [{"ERROR": true, "ERROR_MESSAGE": "..."}]
+
+    Empty payload (e.g. silent rate-limit drop) returns ``[]``.
+    """
+    if not payload or not payload.strip():
+        return []
+    try:
+        data = json.loads(payload)
+    except json.JSONDecodeError as e:
+        logger.warning("AISHub: response is not JSON: %s", e)
+        return []
+    if not isinstance(data, list) or not data:
+        return []
+    header = data[0] if isinstance(data[0], dict) else {}
+    if header.get("ERROR") is True:
+        logger.warning(
+            "AISHub: upstream error: %s",
+            header.get("ERROR_MESSAGE", "<unspecified>"),
+        )
+        return []
+    if len(data) < 2 or not isinstance(data[1], list):
+        return []
+    return [row for row in data[1] if isinstance(row, dict)]
+
+
+def _normalize_record(row: dict) -> dict | None:
+    """Map an AISHub vessel record to our internal vessel schema.
+
+    Returns None when the record can't be used (no MMSI, bad position,
+    sentinel "not available" lat/lng).
+    """
+    try:
+        mmsi = int(row.get("MMSI") or 0)
+    except (TypeError, ValueError):
+        return None
+    if not mmsi:
+        return None
+    try:
+        lat = float(row.get("LATITUDE"))
+        lng = float(row.get("LONGITUDE"))
+    except (TypeError, ValueError):
+        return None
+    # AIS uses 91/181 as "no position available" sentinels.
+    if abs(lat) > 90 or abs(lng) > 180:
+        return None
+    if lat == 91.0 or lng == 181.0:
+        return None
+    # SOG raw 102.3 is "speed not available"; sanitize to 0.
+    try:
+        sog_raw = float(row.get("SOG") or 0)
+    except (TypeError, ValueError):
+        sog_raw = 0.0
+    sog = 0.0 if sog_raw >= 102.2 else sog_raw
+    try:
+        cog = float(row.get("COG") or 0)
+    except (TypeError, ValueError):
+        cog = 0.0
+    try:
+        heading_raw = int(row.get("HEADING") or 511)
+    except (TypeError, ValueError):
+        heading_raw = 511
+    # AIS heading sentinel 511 = "not available" — fall back to COG.
+    heading = heading_raw if heading_raw != 511 else cog
+    try:
+        ais_type = int(row.get("TYPE") or 0)
+    except (TypeError, ValueError):
+        ais_type = 0
+    return {
+        "mmsi": mmsi,
+        "lat": lat,
+        "lng": lng,
+        "sog": sog,
+        "cog": cog,
+        "heading": heading,
+        "name": str(row.get("NAME") or "").strip() or "UNKNOWN",
+        "callsign": str(row.get("CALLSIGN") or "").strip(),
+        "destination": str(row.get("DEST") or "").strip().replace("@", "") or "",
+        "imo": int(row.get("IMO") or 0),
+        "ais_type_code": ais_type,
+    }
+
+
+def fetch_aishub_vessels() -> int:
+    """Poll AISHub and merge vessels into the shared ``_vessels`` store.
+
+    Returns the number of vessels updated (0 on skip, error, or no data).
+    Designed to be called by the APScheduler tier — see
+    ``data_fetcher.py`` for the 20-minute interval job that wraps this.
+    """
+    if not _should_run_fallback():
+        logger.debug("AISHub fallback skipped: primary connected or not configured")
+        return 0
+
+    username = aishub_username()
+    url = (
+        f"{AISHUB_URL}?username={username}&format=1&output=json"
+        f"&compress=0"
+    )
+
+    try:
+        response = fetch_with_curl(url, timeout=30)
+    except Exception as e:
+        logger.warning("AISHub fetch failed: %s", e)
+        return 0
+
+    if not response or response.status_code != 200:
+        logger.warning(
+            "AISHub HTTP %s",
+            getattr(response, "status_code", "None"),
+        )
+        return 0
+
+    rows = _parse_aishub_response(getattr(response, "text", "") or "")
+    if not rows:
+        return 0
+
+    # Inline imports to avoid a circular dependency at module load time
+    # (ais_stream imports lots of things and is loaded by main.py).
+    from services.ais_stream import (
+        _vessels,
+        _vessels_lock,
+        _record_vessel_trail_locked,
+        classify_vessel,
+        get_country_from_mmsi,
+    )
+
+    now = time.time()
+    count = 0
+    with _vessels_lock:
+        for row in rows:
+            normalized = _normalize_record(row)
+            if normalized is None:
+                continue
+            mmsi = normalized["mmsi"]
+            vessel = _vessels.setdefault(mmsi, {"mmsi": mmsi})
+            # Don't overwrite fresher live data: if the WebSocket pushed an
+            # update for this MMSI more recently than now-1s (race during
+            # the brief reconnection window) keep the live one.
+            last = float(vessel.get("_updated") or 0)
+            if last > now - 1:
+                continue
+            vessel.update(
+                {
+                    "lat": normalized["lat"],
+                    "lng": normalized["lng"],
+                    "sog": normalized["sog"],
+                    "cog": normalized["cog"],
+                    "heading": normalized["heading"],
+                    "_updated": now,
+                    "source": "aishub",
+                }
+            )
+            if normalized["name"] and normalized["name"] != "UNKNOWN":
+                vessel["name"] = normalized["name"]
+            if normalized["callsign"]:
+                vessel["callsign"] = normalized["callsign"]
+            if normalized["destination"]:
+                vessel["destination"] = normalized["destination"]
+            if normalized["imo"]:
+                vessel["imo"] = normalized["imo"]
+            if normalized["ais_type_code"]:
+                vessel["ais_type_code"] = normalized["ais_type_code"]
+                vessel["type"] = classify_vessel(normalized["ais_type_code"], mmsi)
+            if not vessel.get("country"):
+                vessel["country"] = get_country_from_mmsi(mmsi)
+            _record_vessel_trail_locked(
+                mmsi,
+                normalized["lat"],
+                normalized["lng"],
+                normalized["sog"],
+                now,
+            )
+            count += 1
+
+    if count:
+        logger.info(
+            "AISHub fallback: merged %d vessels (poll interval %d min)",
+            count,
+            aishub_poll_interval_minutes(),
+        )
+    return count

backend/services/fetchers/earth_observation.py

@@ -1383,10 +1383,21 @@ def _build_uap_sightings_from_hf_mirror() -> list[dict]:
     This is a resilience fallback for local/Windows runs where nuforc.org is
     Cloudflare-gated and the Mapbox token is not configured. It is not as fresh
     as the live NUFORC AJAX feed, but it keeps the layer visible and cached.
+
+    Date-cutoff guard: the kcimc/NUFORC HF dataset is a static snapshot whose
+    maintainer refreshes it sporadically. Without a cutoff, sorting by
+    occurred-desc and taking the top N rows returns whatever the mirror's
+    newest rows happen to be — which can be years old if the snapshot is
+    stale. We apply the same ``_NUFORC_RECENT_DAYS`` window the live path
+    uses (60 days). If the HF mirror has nothing inside the window we return
+    ``[]`` rather than silently serving 3-year-old "newest" rows.
     """
     from services.fetchers.nuforc_enrichment import _HF_CSV_URL, _parse_date
     from services.geocode_validate import coord_in_country
 
+    cutoff_dt = datetime.utcnow() - timedelta(days=_NUFORC_RECENT_DAYS)
+    cutoff_str = cutoff_dt.strftime("%Y-%m-%d")
+
     try:
         response = fetch_with_curl(_HF_CSV_URL, timeout=180, follow_redirects=True)
         if not response or response.status_code != 200:
@@ -1400,6 +1411,7 @@ def _build_uap_sightings_from_hf_mirror() -> list[dict]:
         return []
 
     candidates: list[dict] = []
+    stale_rows_dropped = 0
     try:
         reader = csv.DictReader(io.StringIO(response.text))
         for row in reader:
@@ -1410,6 +1422,9 @@ def _build_uap_sightings_from_hf_mirror() -> list[dict]:
             )
             if not occurred:
                 continue
+            if occurred < cutoff_str:
+                stale_rows_dropped += 1
+                continue
             raw_location = _normalize_uap_location(
                 row.get("Location", "")
                 or row.get("City", "")
@@ -1444,6 +1459,19 @@ def _build_uap_sightings_from_hf_mirror() -> list[dict]:
         logger.warning("UAP sightings: HF fallback parse failed: %s", e)
         return []
 
+    if not candidates:
+        # HF mirror returned rows, but none inside the rolling window. This is
+        # the smoking gun for "the public HF dataset hasn't been refreshed in
+        # years" — log loudly so the operator sees it instead of guessing.
+        logger.error(
+            "UAP sightings: HF fallback yielded 0 rows within last %d days "
+            "(dropped %d stale rows). HF mirror is likely stale; the layer "
+            "will be empty until the live NUFORC path recovers.",
+            _NUFORC_RECENT_DAYS,
+            stale_rows_dropped,
+        )
+        return []
+
     candidates.sort(key=lambda row: (row["occurred"], row["posted"], row["id"]), reverse=True)
     candidates = candidates[:_NUFORC_HF_FALLBACK_LIMIT]
 
@@ -1515,13 +1543,29 @@ def fetch_uap_sightings(*, force_refresh: bool = False):
 
     sightings = _load_nuforc_sightings_cache(force_refresh=force_refresh)
     if sightings is None:
+        live_error: Exception | None = None
         try:
             sightings = _build_recent_uap_sightings()
         except Exception as e:
+            live_error = e
             logger.warning("UAP sightings: live NUFORC rebuild failed, using fallback: %s", e)
             sightings = _build_uap_sightings_from_hf_mirror()
         if sightings:
             _save_nuforc_sightings_cache(sightings)
+        elif live_error is not None:
+            # Both paths failed: live raised AND HF fallback returned empty
+            # (either the HF mirror is stale beyond the cutoff or the network
+            # is gone entirely). The previous code silently set the layer to
+            # ``[]`` and kept marking it fresh; that masked the failure for
+            # days. Surface it via assert_canary so the health registry shows
+            # the layer as broken instead of "fresh and empty".
+            from services.slo import assert_canary
+            assert_canary("uap_sightings", 0)
+            logger.error(
+                "UAP sightings: both live NUFORC and HF fallback produced 0 "
+                "rows; layer is unavailable. Live error: %s",
+                live_error,
+            )
 
     with _data_lock:
         latest_data["uap_sightings"] = sightings or []

backend/services/fetchers/flight_observations.py

@@ -0,0 +1,148 @@
+"""Per-aircraft observation tracking for cumulative fuel/CO2 estimates.
+
+Background
+----------
+The pre-existing emissions enrichment attached a *rate* to each flight
+(GPH and kg/hr) based on aircraft model. Users — reasonably — wanted the
+running total: how much fuel HAS this plane burned since we started
+seeing it? Multiplying the rate by elapsed observation time gets us
+there, but it requires somewhere to remember "when did this icao24
+first appear on our radar?"
+
+Why this lives outside ``flight_trails``
+----------------------------------------
+``flight_trails`` is sized and pruned aggressively for map rendering
+(5-minute TTL for untracked aircraft, 200 trail points max). That's
+wrong for cumulative burn: if a plane has been airborne 2 hours but
+its trail was pruned 30 min in, the "first trail point" timestamp is
+30 min ago, not 2h ago. Worse, when the trail expires and re-creates,
+the cumulative counter would reset mid-flight.
+
+This module tracks observation lifecycle separately:
+
+* When a hex is first observed: start a new flight session.
+* While observed regularly (gap < ``REOPEN_GAP_S``): keep accumulating.
+* When unseen for longer than ``REOPEN_GAP_S``: treat next sighting as
+  a new session (the plane landed and took off again, or it's a
+  different leg). Reset ``first_seen_at``.
+* Stale sessions are pruned every ``PRUNE_INTERVAL_S`` so memory stays
+  bounded.
+
+The user explicitly asked for this counting semantic: "as soon as a
+plane appears there should be a counter that keeps a running count of
+the fuel being burned... If there is no estimate take off time then it
+can just be from the time the server starts to keep a log of whats in
+the air."
+"""
+
+from __future__ import annotations
+
+import threading
+import time
+
+
+# Gap between sightings that resets the session. ADS-B refreshes the
+# whole aircraft list every minute or two, so anything over a few
+# minutes means the plane left our coverage window (landed, transit
+# through dead zone, etc). 15 minutes is conservative.
+REOPEN_GAP_S = 15 * 60
+
+# Don't accumulate runaway memory: drop entries unseen for an hour.
+PRUNE_AFTER_S = 60 * 60
+
+# Cap on accumulated airtime per session so a single bug elsewhere
+# (e.g. ts clock skew) can't produce comically large numbers.
+MAX_SESSION_SECONDS = 24 * 3600  # 24h — longest realistic civilian leg
+
+
+_observations: dict[str, dict[str, float]] = {}
+_lock = threading.Lock()
+_last_prune_at = 0.0
+
+
+def record_observation(icao_hex: str, *, now: float | None = None) -> int:
+    """Record a sighting of ``icao_hex`` and return airtime so far (seconds).
+
+    Returns 0 for the first-ever sighting (no elapsed time yet) or when
+    ``icao_hex`` is falsy. The caller can multiply the returned seconds
+    by ``rate_per_hour / 3600`` to get cumulative consumption.
+    """
+    if not icao_hex:
+        return 0
+    key = str(icao_hex).strip().lower()
+    if not key:
+        return 0
+    current = float(now if now is not None else time.time())
+
+    with _lock:
+        entry = _observations.get(key)
+        if entry is None:
+            _observations[key] = {"first_seen_at": current, "last_seen_at": current}
+            return 0
+        # Use explicit ``is None`` checks instead of ``or`` short-circuit:
+        # ``0.0`` is a legitimate timestamp value (e.g. test fixtures
+        # seeding a far-past first_seen_at to exercise the clamp) but
+        # ``0.0 or fallback`` collapses to ``fallback`` because 0.0 is
+        # falsy. Bit me on my own test — leaving the safer form here.
+        last_raw = entry.get("last_seen_at")
+        last_seen = float(last_raw) if last_raw is not None else current
+        gap = current - last_seen
+        if gap > REOPEN_GAP_S:
+            # Treat as a new flight session — the plane landed/disappeared
+            # long enough that the prior cumulative count is no longer
+            # the same flight.
+            _observations[key] = {"first_seen_at": current, "last_seen_at": current}
+            return 0
+        first_raw = entry.get("first_seen_at")
+        first = float(first_raw) if first_raw is not None else current
+        # Clamp absurd values from clock skew or bad input.
+        elapsed = max(0, min(int(current - first), MAX_SESSION_SECONDS))
+        entry["last_seen_at"] = current
+        return elapsed
+
+
+def prune(*, now: float | None = None) -> int:
+    """Drop entries we haven't seen in ``PRUNE_AFTER_S`` seconds.
+
+    Returns number of entries dropped. Safe to call from a scheduler tick;
+    cheap (single dict scan) so cadence doesn't matter much.
+    """
+    current = float(now if now is not None else time.time())
+    dropped = 0
+    with _lock:
+        stale_keys = []
+        for k, v in _observations.items():
+            last_raw = v.get("last_seen_at")
+            last = float(last_raw) if last_raw is not None else 0.0
+            if current - last > PRUNE_AFTER_S:
+                stale_keys.append(k)
+        for k in stale_keys:
+            del _observations[k]
+            dropped += 1
+    return dropped
+
+
+def get_session_seconds(icao_hex: str, *, now: float | None = None) -> int:
+    """Read-only accessor: airtime for a known icao without bumping last-seen.
+
+    Used by tests and external consumers (e.g. when rendering a snapshot
+    of all in-flight aircraft, you want the current value, not to update
+    last_seen_at as a side effect).
+    """
+    if not icao_hex:
+        return 0
+    key = str(icao_hex).strip().lower()
+    with _lock:
+        entry = _observations.get(key)
+        if entry is None:
+            return 0
+        current = float(now if now is not None else time.time())
+        first_raw = entry.get("first_seen_at")
+        first = float(first_raw) if first_raw is not None else current
+        return max(0, min(int(current - first), MAX_SESSION_SECONDS))
+
+
+def _reset_for_tests() -> None:
+    """Drop all observations. Test helper only."""
+    with _lock:
+        _observations.clear()

backend/services/fetchers/flights.py

@@ -17,6 +17,7 @@ from services.network_utils import fetch_with_curl
 from services.fetchers._store import latest_data, _data_lock, _mark_fresh
 from services.fetchers.plane_alert import enrich_with_plane_alert, enrich_with_tracked_names
 from services.fetchers.emissions import get_emissions_info
+from services.fetchers.flight_observations import record_observation as _record_flight_observation
 from services.fetchers.retry import with_retry
 from services.fetchers.route_database import lookup_route
 from services.fetchers.aircraft_database import lookup_aircraft_type
@@ -29,6 +30,88 @@ _RE_AIRLINE_CODE_1 = re.compile(r"^([A-Z]{3})\d")
 _RE_AIRLINE_CODE_2 = re.compile(r"^([A-Z]{3})[A-Z\d]")
 
 
+def detect_gps_jamming_zones(
+    raw_flights: list[dict],
+    *,
+    min_aircraft: int | None = None,
+    min_ratio: float | None = None,
+    nacp_threshold: int | None = None,
+) -> list[dict]:
+    """Detect GPS interference zones from a snapshot of raw ADS-B aircraft.
+
+    Methodology mirrors GPSJam.org / Flightradar24: bin aircraft into 1°x1°
+    grid cells, flag cells where the fraction of aircraft reporting degraded
+    NACp clears a threshold.
+
+    Inputs
+    ------
+    raw_flights:
+        Iterable of dicts. Each item is expected to carry ``lat``, ``lng``
+        (or ``lon``), and ``nac_p``. Records missing position OR missing
+        ``nac_p`` entirely (typical for OpenSky-sourced flights) are
+        skipped — absence-of-data isn't evidence of anything.
+
+    nac_p == 0 IS counted as degraded. Pre-fix code skipped it on the theory
+    that "0 = old transponder, never computed accuracy." That's only half
+    right: modern Mode-S Enhanced Surveillance transponders also fall back
+    to nac_p=0 when they lose GPS lock entirely — which is exactly the
+    jamming signature we're trying to detect. Filtering 0 out was discarding
+    the strongest evidence.
+
+    Denoising:
+        1. Require ``min_aircraft`` per grid cell for statistical validity.
+        2. Subtract 1 from degraded count per cell (GPSJam's technique) so
+           a single quirky transponder can't flag an entire zone.
+        3. Require ratio ``adjusted_degraded / total > min_ratio``.
+
+    All thresholds default to the module-level constants but can be
+    overridden for testing.
+    """
+    min_aircraft = GPS_JAMMING_MIN_AIRCRAFT if min_aircraft is None else int(min_aircraft)
+    min_ratio = GPS_JAMMING_MIN_RATIO if min_ratio is None else float(min_ratio)
+    nacp_threshold = (
+        GPS_JAMMING_NACP_THRESHOLD if nacp_threshold is None else int(nacp_threshold)
+    )
+
+    jamming_grid: dict[str, dict[str, int]] = {}
+    for rf in raw_flights or []:
+        rlat = rf.get("lat")
+        rlng = rf.get("lng") if rf.get("lng") is not None else rf.get("lon")
+        if rlat is None or rlng is None:
+            continue
+        nacp = rf.get("nac_p")
+        if nacp is None:
+            continue
+        grid_key = f"{int(rlat)},{int(rlng)}"
+        cell = jamming_grid.setdefault(grid_key, {"degraded": 0, "total": 0})
+        cell["total"] += 1
+        if nacp < nacp_threshold:
+            cell["degraded"] += 1
+
+    jamming_zones: list[dict] = []
+    for gk, counts in jamming_grid.items():
+        if counts["total"] < min_aircraft:
+            continue
+        adjusted_degraded = max(counts["degraded"] - 1, 0)
+        if adjusted_degraded == 0:
+            continue
+        ratio = adjusted_degraded / counts["total"]
+        if ratio > min_ratio:
+            lat_i, lng_i = gk.split(",")
+            severity = "low" if ratio < 0.5 else "medium" if ratio < 0.75 else "high"
+            jamming_zones.append(
+                {
+                    "lat": int(lat_i) + 0.5,
+                    "lng": int(lng_i) + 0.5,
+                    "severity": severity,
+                    "ratio": round(ratio, 2),
+                    "degraded": counts["degraded"],
+                    "total": counts["total"],
+                }
+            )
+    return jamming_zones
+
+
 # ---------------------------------------------------------------------------
 # OpenSky Network API Client (OAuth2)
 # ---------------------------------------------------------------------------
@@ -519,6 +602,22 @@ def _classify_and_publish(all_adsb_flights):
         if model:
             emi = get_emissions_info(model)
             if emi:
+                # Cumulative fuel/CO2: multiply the per-hour rate by how
+                # long we've been observing this airframe. Users want to
+                # see the *amount* burned, not just the rate. If we've
+                # never seen this hex before, observed_seconds is 0 and
+                # the cumulative values are 0 until the next refresh —
+                # the rate is still useful info on its own.
+                observed_seconds = _record_flight_observation(
+                    f.get("icao24") or ""
+                )
+                elapsed_h = observed_seconds / 3600.0
+                emi = {
+                    **emi,
+                    "observed_seconds": observed_seconds,
+                    "fuel_gallons_burned": round(emi["fuel_gph"] * elapsed_h, 1),
+                    "co2_kg_emitted": round(emi["co2_kg_per_hour"] * elapsed_h, 1),
+                }
                 f["emissions"] = emi
 
         callsign = f.get("callsign", "").strip().upper()
@@ -737,56 +836,8 @@ def _classify_and_publish(all_adsb_flights):
         latest_data["military_flights"] = military_snapshot
 
     # --- GPS Jamming Detection ---
-    # Uses NACp (Navigation Accuracy Category – Position) from ADS-B to infer
-    # GPS interference zones, similar to GPSJam.org / Flightradar24.
-    # NACp < 8 = position accuracy worse than the FAA-mandated 0.05 NM.
-    #
-    # Denoising (to suppress false positives from old GA transponders):
-    # 1. Skip nac_p == 0 ("unknown accuracy") — old transponders that never
-    #    computed accuracy, NOT evidence of jamming.  Real jamming shows 1-7.
-    # 2. Require minimum aircraft per grid cell for statistical validity.
-    # 3. Subtract 1 from degraded count per cell (GPSJam's technique) so a
-    #    single quirky transponder can't flag an entire zone.
-    # 4. Require the adjusted ratio to exceed the threshold.
     try:
-        jamming_grid = {}
-        raw_flights = raw_flights_snapshot
-        for rf in raw_flights:
-            rlat = rf.get("lat")
-            rlng = rf.get("lng") or rf.get("lon")
-            if rlat is None or rlng is None:
-                continue
-            nacp = rf.get("nac_p")
-            if nacp is None or nacp == 0:
-                continue
-            grid_key = f"{int(rlat)},{int(rlng)}"
-            if grid_key not in jamming_grid:
-                jamming_grid[grid_key] = {"degraded": 0, "total": 0}
-            jamming_grid[grid_key]["total"] += 1
-            if nacp < GPS_JAMMING_NACP_THRESHOLD:
-                jamming_grid[grid_key]["degraded"] += 1
-
-        jamming_zones = []
-        for gk, counts in jamming_grid.items():
-            if counts["total"] < GPS_JAMMING_MIN_AIRCRAFT:
-                continue
-            adjusted_degraded = max(counts["degraded"] - 1, 0)
-            if adjusted_degraded == 0:
-                continue
-            ratio = adjusted_degraded / counts["total"]
-            if ratio > GPS_JAMMING_MIN_RATIO:
-                lat_i, lng_i = gk.split(",")
-                severity = "low" if ratio < 0.5 else "medium" if ratio < 0.75 else "high"
-                jamming_zones.append(
-                    {
-                        "lat": int(lat_i) + 0.5,
-                        "lng": int(lng_i) + 0.5,
-                        "severity": severity,
-                        "ratio": round(ratio, 2),
-                        "degraded": counts["degraded"],
-                        "total": counts["total"],
-                    }
-                )
+        jamming_zones = detect_gps_jamming_zones(raw_flights_snapshot)
         with _data_lock:
             latest_data["gps_jamming"] = jamming_zones
         if jamming_zones:

backend/services/fetchers/military.py

@@ -7,6 +7,7 @@ import requests
 from services.network_utils import fetch_with_curl
 from services.fetchers._store import latest_data, _data_lock, _mark_fresh
 from services.fetchers.emissions import get_emissions_info
+from services.fetchers.flight_observations import record_observation as _record_flight_observation
 from services.fetchers.plane_alert import enrich_with_plane_alert
 
 logger = logging.getLogger("services.data_fetcher")
@@ -300,6 +301,18 @@ def fetch_military_flights():
         if model:
             emissions = get_emissions_info(model)
             if emissions:
+                # Cumulative fuel/CO2 since first observation — mirrors
+                # the civilian path in flights._classify_and_publish.
+                observed_seconds = _record_flight_observation(
+                    mf.get("icao24") or ""
+                )
+                elapsed_h = observed_seconds / 3600.0
+                emissions = {
+                    **emissions,
+                    "observed_seconds": observed_seconds,
+                    "fuel_gallons_burned": round(emissions["fuel_gph"] * elapsed_h, 1),
+                    "co2_kg_emitted": round(emissions["co2_kg_per_hour"] * elapsed_h, 1),
+                }
                 mf["emissions"] = emissions
         if mf.get("alert_category"):
             mf["type"] = "tracked_flight"

backend/services/mesh/mesh_hashchain.py

@@ -33,8 +33,9 @@ Each event contains:
 
 Persistence: JSON file at backend/data/infonet.json
 
-Encrypted gate chat events are intentionally kept off the public chain and
-persisted separately via GateMessageStore.
+Encrypted gate chat events are private-chain ciphertext records. They are
+excluded from public read surfaces and replicated only over private Infonet
+transports.
 """
 
 import json
@@ -64,6 +65,8 @@ from services.mesh.mesh_schema import (
     ACTIVE_PUBLIC_LEDGER_EVENT_TYPES,
     PUBLIC_LEDGER_EVENT_TYPES,
     validate_event_payload,
+    validate_private_dm_ledger_payload,
+    validate_private_gate_ledger_payload,
     validate_protocol_fields,
     validate_public_ledger_payload,
 )
@@ -127,6 +130,12 @@ GATE_SEGMENT_MAX_COMPRESSED_BYTES = max(
     int(os.environ.get("MESH_GATE_SEGMENT_MAX_COMPRESSED_BYTES", str(2 * 1024 * 1024)) or str(2 * 1024 * 1024)),
 )
 GATE_SEGMENT_STORAGE_VERSION = 1
+DM_HASHCHAIN_SPOOL_LIMIT = max(1, int(os.environ.get("MESH_DM_HASHCHAIN_SPOOL_LIMIT", "2") or "2"))
+DM_HASHCHAIN_SPOOL_SENDER_LIMIT = max(
+    1,
+    int(os.environ.get("MESH_DM_HASHCHAIN_SPOOL_SENDER_LIMIT", "1") or "1"),
+)
+DM_HASHCHAIN_SPOOL_TTL_S = max(60, int(os.environ.get("MESH_DM_HASHCHAIN_SPOOL_TTL_S", "3600") or "3600"))
 _PUBLIC_EVENT_APPEND_HOOKS: list[Any] = []
 _PUBLIC_EVENT_APPEND_HOOKS_LOCK = threading.Lock()
 
@@ -340,6 +349,32 @@ def _private_gate_event_id(
     ).hexdigest()
 
 
+def _private_gate_signature_payload_variants(gate_id: str, event: dict[str, Any]) -> list[dict[str, Any]]:
+    payload = _private_gate_signature_payload(gate_id, event)
+    variants: list[dict[str, Any]] = [payload]
+    event_payload = event.get("payload") if isinstance(event.get("payload"), dict) else {}
+    reply_to = str(event_payload.get("reply_to", "") or "").strip()
+    if reply_to:
+        variants.append(_private_gate_signature_payload(gate_id, event, include_reply_to=False))
+    if "epoch" in payload:
+        no_epoch = dict(payload)
+        no_epoch.pop("epoch", None)
+        variants.append(no_epoch)
+        if reply_to:
+            no_epoch_no_reply = _private_gate_signature_payload(gate_id, event, include_reply_to=False)
+            no_epoch_no_reply.pop("epoch", None)
+            variants.append(no_epoch_no_reply)
+    deduped: list[dict[str, Any]] = []
+    seen: set[str] = set()
+    for variant in variants:
+        material = json.dumps(variant, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
+        if material in seen:
+            continue
+        seen.add(material)
+        deduped.append(variant)
+    return deduped
+
+
 def _sanitize_private_gate_event(gate_id: str, event: dict[str, Any]) -> dict[str, Any]:
     payload = event.get("payload") if isinstance(event.get("payload"), dict) else {}
     sanitized = {
@@ -1568,11 +1603,18 @@ class Infonet:
     def _rebuild_state(self) -> None:
         self.event_index = {}
         self.node_sequences = {}
-        # Keep private signed-write replay domains across public-chain
-        # rebuilds; these domains protect local side effects that are not
-        # represented as public Infonet events.
-        if not isinstance(getattr(self, "sequence_domains", None), dict):
-            self.sequence_domains = {}
+        # Keep private signed-write replay domains that are not represented
+        # on-chain, but rebuild the gate_message sequence domain from chain
+        # events so reloads/fork application do not mix it with public
+        # per-node message sequences.
+        preserved_domains = {}
+        if isinstance(getattr(self, "sequence_domains", None), dict):
+            preserved_domains = {
+                key: value
+                for key, value in self.sequence_domains.items()
+                if not str(key or "").endswith("|gate_message")
+            }
+        self.sequence_domains = dict(preserved_domains)
         self.public_key_bindings = {}
         self.revocations = {}
         self._replay_filter = ReplayFilter()
@@ -1584,9 +1626,12 @@ class Infonet:
             node_id = evt.get("node_id", "")
             sequence = _safe_int(evt.get("sequence", 0) or 0, 0)
             if node_id and sequence:
-                last = self.node_sequences.get(node_id, 0)
+                sequence_table, sequence_key = self._sequence_table_for_event(
+                    evt.get("event_type", ""), node_id
+                )
+                last = sequence_table.get(sequence_key, 0)
                 if sequence > last:
-                    self.node_sequences[node_id] = sequence
+                    sequence_table[sequence_key] = sequence
             public_key = str(evt.get("public_key", "") or "")
             if public_key and node_id:
                 existing = self.public_key_bindings.get(public_key)
@@ -1898,6 +1943,295 @@ class Infonet:
         self._save()
         return True, "ok"
 
+    def _sequence_table_for_event(self, event_type: str, node_id: str) -> tuple[dict[str, int], str]:
+        normalized = str(event_type or "").strip().lower()
+        if normalized == "gate_message":
+            return self.sequence_domains, f"{node_id}|gate_message"
+        if normalized == "dm_message":
+            return self.sequence_domains, f"{node_id}|dm_message"
+        return self.node_sequences, node_id
+
+    def _dm_spool_target_key(self, payload: dict[str, Any]) -> tuple[str, str]:
+        delivery_class = str(payload.get("delivery_class", "") or "").strip().lower()
+        if delivery_class == "shared":
+            key = str(payload.get("recipient_token", "") or "").strip()
+        else:
+            key = str(payload.get("recipient_id", "") or "").strip()
+        return delivery_class, key
+
+    def _dm_spool_active_counts(
+        self,
+        payload: dict[str, Any],
+        *,
+        sender_id: str = "",
+        now: float | None = None,
+    ) -> tuple[int, int]:
+        delivery_class, key = self._dm_spool_target_key(payload)
+        if not key:
+            return 0, 0
+        sender_id = str(sender_id or "").strip()
+        current = time.time() if now is None else float(now)
+        total_count = 0
+        sender_count = 0
+        for evt in reversed(self.events):
+            if evt.get("event_type") != "dm_message":
+                continue
+            evt_payload = evt.get("payload") if isinstance(evt.get("payload"), dict) else {}
+            evt_delivery_class, evt_key = self._dm_spool_target_key(evt_payload)
+            if evt_delivery_class != delivery_class:
+                continue
+            if evt_key != key:
+                continue
+            evt_ts = float(evt_payload.get("timestamp", evt.get("timestamp", 0)) or 0)
+            if evt_ts > 0 and current - evt_ts > DM_HASHCHAIN_SPOOL_TTL_S:
+                continue
+            total_count += 1
+            if sender_id and str(evt.get("node_id", "") or "").strip() == sender_id:
+                sender_count += 1
+            if total_count >= DM_HASHCHAIN_SPOOL_LIMIT and (
+                not sender_id or sender_count >= DM_HASHCHAIN_SPOOL_SENDER_LIMIT
+            ):
+                break
+        return total_count, sender_count
+
+    def _dm_spool_active_count(self, payload: dict[str, Any], *, now: float | None = None) -> int:
+        total_count, _sender_count = self._dm_spool_active_counts(payload, now=now)
+        return total_count
+
+    def append_private_dm_message(
+        self,
+        *,
+        node_id: str,
+        payload: dict,
+        signature: str,
+        sequence: int,
+        public_key: str,
+        public_key_algo: str,
+        protocol_version: str = "",
+        timestamp: float = 0,
+    ) -> dict:
+        """Append an encrypted DM dead-drop message to the private Infonet ledger.
+
+        The event is a small offline spool, capped per mailbox target, so the
+        hashchain can carry a couple of sealed DMs without becoming an
+        unbounded global mailbox.
+        """
+        event_type = "dm_message"
+        if sequence <= 0:
+            raise ValueError("sequence is required and must be > 0")
+        sequence_table, sequence_key = self._sequence_table_for_event(event_type, node_id)
+        last = sequence_table.get(sequence_key, 0)
+        if sequence <= last:
+            raise ValueError(f"Replay detected: sequence {sequence} <= last {last}")
+
+        raw_payload = dict(payload or {})
+        if "message" in raw_payload or "plaintext" in raw_payload or "_local_plaintext" in raw_payload:
+            raise ValueError("private DM ledger payload must not contain plaintext")
+        if str(raw_payload.get("transport_lock", "") or "").strip().lower() != "private_strong":
+            raise ValueError("DM hashchain spool requires private_strong transport_lock")
+
+        payload = normalize_payload(event_type, raw_payload)
+        ok, reason = validate_private_dm_ledger_payload(payload)
+        if not ok:
+            raise ValueError(reason)
+        total_count, sender_count = self._dm_spool_active_counts(payload, sender_id=node_id)
+        if sender_count >= DM_HASHCHAIN_SPOOL_SENDER_LIMIT:
+            raise ValueError("DM hashchain sender spool full for recipient")
+        if total_count >= DM_HASHCHAIN_SPOOL_LIMIT:
+            raise ValueError("DM hashchain spool full for recipient")
+
+        payload_json = json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
+        if len(payload_json.encode("utf-8")) > MAX_PAYLOAD_BYTES:
+            raise ValueError("payload exceeds max size")
+
+        protocol_version = str(protocol_version or PROTOCOL_VERSION)
+        ok, reason = validate_protocol_fields(protocol_version, NETWORK_ID)
+        if not ok:
+            raise ValueError(reason)
+
+        if not (signature and public_key and public_key_algo):
+            raise ValueError("Missing signature fields")
+        algo = parse_public_key_algo(public_key_algo)
+        if not algo:
+            raise ValueError("Unsupported public_key_algo")
+        if not verify_node_binding(node_id, public_key):
+            raise ValueError("node_id mismatch")
+        bound, bind_reason = self._bind_public_key(public_key, node_id)
+        if not bound:
+            raise ValueError(bind_reason)
+        sig_payload = build_signature_payload(
+            event_type=event_type,
+            node_id=node_id,
+            sequence=sequence,
+            payload=payload,
+        )
+        if not verify_signature(
+            public_key_b64=public_key,
+            public_key_algo=public_key_algo,
+            signature_hex=signature,
+            payload=sig_payload,
+        ):
+            raise ValueError("Invalid signature")
+
+        revoked, _info = self._revocation_status(public_key)
+        if revoked:
+            raise ValueError("public key is revoked")
+
+        event = ChainEvent(
+            prev_hash=self.head_hash,
+            event_type=event_type,
+            node_id=node_id,
+            payload=payload,
+            timestamp=float(timestamp or time.time()),
+            sequence=sequence,
+            signature=signature,
+            public_key=public_key,
+            public_key_algo=public_key_algo,
+            protocol_version=protocol_version,
+        )
+        event_dict = event.to_dict()
+        self._write_wal(event_dict)
+        self.events.append(event_dict)
+        self.event_index[event.event_id] = len(self.events) - 1
+        self.head_hash = event.event_id
+        sequence_table[sequence_key] = sequence
+        self._replay_filter.add(event.event_id)
+        self._invalidate_merkle_cache()
+        self._update_counters_for_event(event_dict)
+        self._save()
+
+        try:
+            from services.mesh.mesh_rns import rns_bridge
+
+            rns_bridge.publish_event(event_dict)
+        except Exception:
+            pass
+        _notify_public_event_append_hooks(event_dict)
+        logger.info(
+            f"Infonet append [dm_message] by {_redact_node(node_id)} seq={sequence} "
+            f"id={event.event_id[:16]}..."
+        )
+        return event_dict
+
+    def append_private_gate_message(
+        self,
+        *,
+        node_id: str,
+        payload: dict,
+        signature: str,
+        sequence: int,
+        public_key: str,
+        public_key_algo: str,
+        protocol_version: str = "",
+        timestamp: float = 0,
+    ) -> dict:
+        """Append an encrypted gate message to the private Infonet ledger.
+
+        Gate messages use their own sequence domain so a gate post cannot
+        consume or replay-block the author's public broadcast sequence.
+        """
+        event_type = "gate_message"
+        if sequence <= 0:
+            raise ValueError("sequence is required and must be > 0")
+        sequence_table, sequence_key = self._sequence_table_for_event(event_type, node_id)
+        last = sequence_table.get(sequence_key, 0)
+        if sequence <= last:
+            raise ValueError(f"Replay detected: sequence {sequence} <= last {last}")
+
+        raw_payload = dict(payload or {})
+        if "message" in raw_payload or "_local_plaintext" in raw_payload or "_local_reply_to" in raw_payload:
+            raise ValueError("private gate ledger payload must not contain plaintext")
+        if str(raw_payload.get("transport_lock", "") or "").strip().lower() != "private_strong":
+            raise ValueError("gate messages require private_strong transport_lock")
+
+        payload = normalize_payload(event_type, raw_payload)
+        ok, reason = validate_private_gate_ledger_payload(payload)
+        if not ok:
+            raise ValueError(reason)
+
+        payload_json = json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
+        if len(payload_json.encode("utf-8")) > MAX_PAYLOAD_BYTES:
+            raise ValueError("payload exceeds max size")
+
+        protocol_version = str(protocol_version or PROTOCOL_VERSION)
+        ok, reason = validate_protocol_fields(protocol_version, NETWORK_ID)
+        if not ok:
+            raise ValueError(reason)
+
+        if not (signature and public_key and public_key_algo):
+            raise ValueError("Missing signature fields")
+        algo = parse_public_key_algo(public_key_algo)
+        if not algo:
+            raise ValueError("Unsupported public_key_algo")
+        if not verify_node_binding(node_id, public_key):
+            raise ValueError("node_id mismatch")
+        bound, bind_reason = self._bind_public_key(public_key, node_id)
+        if not bound:
+            raise ValueError(bind_reason)
+        event_for_signature = {"payload": payload}
+        signature_ok = False
+        for signature_payload in _private_gate_signature_payload_variants(
+            str(payload.get("gate", "") or ""),
+            event_for_signature,
+        ):
+            sig_payload = build_signature_payload(
+                event_type=event_type,
+                node_id=node_id,
+                sequence=sequence,
+                payload=signature_payload,
+            )
+            if verify_signature(
+                public_key_b64=public_key,
+                public_key_algo=public_key_algo,
+                signature_hex=signature,
+                payload=sig_payload,
+            ):
+                signature_ok = True
+                break
+        if not signature_ok:
+            raise ValueError("Invalid signature")
+
+        revoked, _info = self._revocation_status(public_key)
+        if revoked:
+            raise ValueError("public key is revoked")
+
+        event = ChainEvent(
+            prev_hash=self.head_hash,
+            event_type=event_type,
+            node_id=node_id,
+            payload=payload,
+            timestamp=float(timestamp or time.time()),
+            sequence=sequence,
+            signature=signature,
+            public_key=public_key,
+            public_key_algo=public_key_algo,
+            protocol_version=protocol_version,
+        )
+        event_dict = event.to_dict()
+        self._write_wal(event_dict)
+        self.events.append(event_dict)
+        self.event_index[event.event_id] = len(self.events) - 1
+        self.head_hash = event.event_id
+        sequence_table[sequence_key] = sequence
+        self._replay_filter.add(event.event_id)
+        self._invalidate_merkle_cache()
+        self._update_counters_for_event(event_dict)
+        self._save()
+
+        try:
+            from services.mesh.mesh_rns import rns_bridge
+
+            rns_bridge.publish_event(event_dict)
+        except Exception:
+            pass
+        _notify_public_event_append_hooks(event_dict)
+
+        logger.info(
+            f"Infonet append [gate_message] by {_redact_node(node_id)} seq={sequence} "
+            f"id={event.event_id[:16]}..."
+        )
+        return event_dict
+
     def append(
         self,
         event_type: str,
@@ -2078,6 +2412,18 @@ class Infonet:
             if not event_id or not prev_hash:
                 rejected.append({"index": idx, "reason": "Missing event_id or prev_hash"})
                 continue
+            if event_id in self.event_index:
+                duplicates += 1
+                continue
+            if self._replay_filter.seen(event_id):
+                try:
+                    from services.mesh.mesh_metrics import increment as metrics_inc
+
+                    metrics_inc("ingest_replay_seen")
+                except Exception:
+                    pass
+                duplicates += 1
+                continue
             if prev_hash != expected_prev:
                 try:
                     from services.mesh.mesh_metrics import increment as metrics_inc
@@ -2096,25 +2442,14 @@ class Infonet:
                     pass
                 rejected.append({"index": idx, "reason": "network_id mismatch"})
                 continue
-            if event_id in self.event_index:
-                duplicates += 1
-                continue
-            if self._replay_filter.seen(event_id):
-                try:
-                    from services.mesh.mesh_metrics import increment as metrics_inc
-
-                    metrics_inc("ingest_replay_seen")
-                except Exception:
-                    pass
-                duplicates += 1
-                continue
             if prev_hash != self.head_hash:
                 rejected.append({"index": idx, "reason": "prev_hash does not match head"})
                 continue
             if sequence <= 0:
                 rejected.append({"index": idx, "reason": "Invalid sequence"})
                 continue
-            last = self.node_sequences.get(node_id, 0)
+            sequence_table, sequence_key = self._sequence_table_for_event(event_type, node_id)
+            last = sequence_table.get(sequence_key, 0)
             if sequence <= last:
                 rejected.append({"index": idx, "reason": "Replay detected"})
                 continue
@@ -2149,7 +2484,18 @@ class Infonet:
             if not ok:
                 rejected.append({"index": idx, "reason": reason})
                 continue
-            ok, reason = validate_public_ledger_payload(event_type, payload)
+            if event_type == "gate_message":
+                ok, reason = validate_private_gate_ledger_payload(payload)
+            elif event_type == "dm_message":
+                ok, reason = validate_private_dm_ledger_payload(payload)
+                if ok:
+                    total_count, sender_count = self._dm_spool_active_counts(payload, sender_id=str(evt.get("node_id", "") or ""))
+                    if sender_count >= DM_HASHCHAIN_SPOOL_SENDER_LIMIT:
+                        ok, reason = False, "DM hashchain sender spool full for recipient"
+                    elif total_count >= DM_HASHCHAIN_SPOOL_LIMIT:
+                        ok, reason = False, "DM hashchain spool full for recipient"
+            else:
+                ok, reason = validate_public_ledger_payload(event_type, payload)
             if not ok:
                 rejected.append({"index": idx, "reason": reason})
                 continue
@@ -2225,7 +2571,7 @@ class Infonet:
                     pass
                 rejected.append({"index": idx, "reason": "public key is revoked"})
                 continue
-            last_seq = self.node_sequences.get(node_id, 0)
+            last_seq = sequence_table.get(sequence_key, 0)
             if sequence <= last_seq:
                 try:
                     from services.mesh.mesh_metrics import increment as metrics_inc
@@ -2261,18 +2607,30 @@ class Infonet:
                 rejected.append({"index": idx, "reason": bind_reason})
                 continue
 
-            sig_payload = build_signature_payload(
-                event_type=event_type,
-                node_id=node_id,
-                sequence=sequence,
-                payload=payload,
-            )
-            if not verify_signature(
-                public_key_b64=public_key,
-                public_key_algo=public_key_algo,
-                signature_hex=signature,
-                payload=sig_payload,
-            ):
+            if event_type == "gate_message":
+                signature_payloads = _private_gate_signature_payload_variants(
+                    str(payload.get("gate", "") or ""),
+                    evt,
+                )
+            else:
+                signature_payloads = [payload]
+            signature_ok = False
+            for signature_payload in signature_payloads:
+                sig_payload = build_signature_payload(
+                    event_type=event_type,
+                    node_id=node_id,
+                    sequence=sequence,
+                    payload=signature_payload,
+                )
+                if verify_signature(
+                    public_key_b64=public_key,
+                    public_key_algo=public_key_algo,
+                    signature_hex=signature,
+                    payload=sig_payload,
+                ):
+                    signature_ok = True
+                    break
+            if not signature_ok:
                 try:
                     from services.mesh.mesh_metrics import increment as metrics_inc
 
@@ -2302,7 +2660,7 @@ class Infonet:
             self.events.append(evt)
             self.event_index[event_id] = len(self.events) - 1
             self.head_hash = event_id
-            self.node_sequences[node_id] = sequence
+            sequence_table[sequence_key] = sequence
             self._update_counters_for_event(evt)
             accepted += 1
             expected_prev = event_id
@@ -2365,6 +2723,7 @@ class Infonet:
                     verify_node_binding,
                 )
 
+                event_type = evt_dict.get("event_type", "")
                 node_id = evt_dict.get("node_id", "")
                 if not parse_public_key_algo(public_key_algo):
                     return False, f"Unsupported public_key_algo at index {i}"
@@ -2375,21 +2734,41 @@ class Infonet:
                     return False, f"public key binding conflict at index {i}"
                 seen_public_keys[public_key] = node_id
 
-                normalized = normalize_payload(
-                    evt_dict.get("event_type", ""), evt_dict.get("payload", {})
-                )
-                sig_payload = build_signature_payload(
-                    event_type=evt_dict.get("event_type", ""),
-                    node_id=node_id,
-                    sequence=_safe_int(evt_dict.get("sequence", 0) or 0, 0),
-                    payload=normalized,
-                )
-                if not verify_signature(
-                    public_key_b64=public_key,
-                    public_key_algo=public_key_algo,
-                    signature_hex=signature,
-                    payload=sig_payload,
-                ):
+                payload = evt_dict.get("payload", {})
+                if event_type == "gate_message":
+                    ok, reason = validate_private_gate_ledger_payload(payload)
+                    if not ok:
+                        return False, f"Invalid gate_message payload at index {i}: {reason}"
+                    signature_payloads = _private_gate_signature_payload_variants(
+                        str(payload.get("gate", "") or ""),
+                        evt_dict,
+                    )
+                elif event_type == "dm_message":
+                    ok, reason = validate_private_dm_ledger_payload(payload)
+                    if not ok:
+                        return False, f"Invalid dm_message payload at index {i}: {reason}"
+                    signature_payloads = [normalize_payload(event_type, payload)]
+                else:
+                    signature_payloads = [
+                        normalize_payload(event_type, payload)
+                    ]
+                signature_ok = False
+                for signature_payload in signature_payloads:
+                    sig_payload = build_signature_payload(
+                        event_type=event_type,
+                        node_id=node_id,
+                        sequence=_safe_int(evt_dict.get("sequence", 0) or 0, 0),
+                        payload=signature_payload,
+                    )
+                    if verify_signature(
+                        public_key_b64=public_key,
+                        public_key_algo=public_key_algo,
+                        signature_hex=signature,
+                        payload=sig_payload,
+                    ):
+                        signature_ok = True
+                        break
+                if not signature_ok:
                     return False, f"Invalid signature at index {i}"
 
             prev = evt_dict["event_id"]
@@ -2454,27 +2833,48 @@ class Infonet:
                     verify_node_binding,
                 )
 
+                event_type = evt_dict.get("event_type", "")
                 node_id = evt_dict.get("node_id", "")
                 if not parse_public_key_algo(public_key_algo):
                     return False, f"Unsupported public_key_algo at index {i}"
                 if not verify_node_binding(node_id, public_key):
                     return False, f"node_id mismatch at index {i}"
 
-                normalized = normalize_payload(
-                    evt_dict.get("event_type", ""), evt_dict.get("payload", {})
-                )
-                sig_payload = build_signature_payload(
-                    event_type=evt_dict.get("event_type", ""),
-                    node_id=node_id,
-                    sequence=_safe_int(evt_dict.get("sequence", 0) or 0, 0),
-                    payload=normalized,
-                )
-                if not verify_signature(
-                    public_key_b64=public_key,
-                    public_key_algo=public_key_algo,
-                    signature_hex=signature,
-                    payload=sig_payload,
-                ):
+                payload = evt_dict.get("payload", {})
+                if event_type == "gate_message":
+                    ok, reason = validate_private_gate_ledger_payload(payload)
+                    if not ok:
+                        return False, f"Invalid gate_message payload at index {i}: {reason}"
+                    signature_payloads = _private_gate_signature_payload_variants(
+                        str(payload.get("gate", "") or ""),
+                        evt_dict,
+                    )
+                elif event_type == "dm_message":
+                    ok, reason = validate_private_dm_ledger_payload(payload)
+                    if not ok:
+                        return False, f"Invalid dm_message payload at index {i}: {reason}"
+                    signature_payloads = [normalize_payload(event_type, payload)]
+                else:
+                    signature_payloads = [
+                        normalize_payload(event_type, payload)
+                    ]
+                signature_ok = False
+                for signature_payload in signature_payloads:
+                    sig_payload = build_signature_payload(
+                        event_type=event_type,
+                        node_id=node_id,
+                        sequence=_safe_int(evt_dict.get("sequence", 0) or 0, 0),
+                        payload=signature_payload,
+                    )
+                    if verify_signature(
+                        public_key_b64=public_key,
+                        public_key_algo=public_key_algo,
+                        signature_hex=signature,
+                        payload=sig_payload,
+                    ):
+                        signature_ok = True
+                        break
+                if not signature_ok:
                     return False, f"Invalid signature at index {i}"
             prev = evt_dict["event_id"]
 
@@ -2538,7 +2938,14 @@ class Infonet:
             node_id = evt.get("node_id", "")
             sequence = _safe_int(evt.get("sequence", 0) or 0, 0)
             if node_id and sequence:
-                last_seq[node_id] = max(last_seq.get(node_id, 0), sequence)
+                sequence_key = (
+                    f"{node_id}|gate_message"
+                    if str(evt.get("event_type", "") or "").strip().lower() == "gate_message"
+                    else f"{node_id}|dm_message"
+                    if str(evt.get("event_type", "") or "").strip().lower() == "dm_message"
+                    else node_id
+                )
+                last_seq[sequence_key] = max(last_seq.get(sequence_key, 0), sequence)
             public_key = str(evt.get("public_key", "") or "")
             if public_key and node_id:
                 seen_public_keys.setdefault(public_key, node_id)
@@ -2558,8 +2965,21 @@ class Infonet:
             existing_idx = self.event_index.get(event_id)
             if existing_idx is not None and existing_idx <= prev_index:
                 return False, "duplicate event_id"
-            payload = normalize_payload(event_type, dict(payload or {}))
+            if event_type == "gate_message":
+                payload = dict(payload or {})
+            elif event_type == "dm_message":
+                payload = normalize_payload(event_type, dict(payload or {}))
+            else:
+                payload = normalize_payload(event_type, dict(payload or {}))
             ok, reason = validate_event_payload(event_type, payload)
+            if not ok:
+                return False, reason
+            if event_type == "gate_message":
+                ok, reason = validate_private_gate_ledger_payload(payload)
+            elif event_type == "dm_message":
+                ok, reason = validate_private_dm_ledger_payload(payload)
+            else:
+                ok, reason = validate_public_ledger_payload(event_type, payload)
             if not ok:
                 return False, reason
             proto = evt.get("protocol_version") or PROTOCOL_VERSION
@@ -2573,7 +2993,14 @@ class Infonet:
             revoked, _info = self._revocation_status(public_key)
             if revoked and event_type != "key_revoke":
                 return False, "public key revoked"
-            last = last_seq.get(node_id, 0)
+            sequence_key = (
+                f"{node_id}|gate_message"
+                if event_type == "gate_message"
+                else f"{node_id}|dm_message"
+                if event_type == "dm_message"
+                else node_id
+            )
+            last = last_seq.get(sequence_key, 0)
             if sequence <= last:
                 return False, "sequence replay"
             from services.mesh.mesh_crypto import (
@@ -2591,23 +3018,35 @@ class Infonet:
             if existing and existing != node_id:
                 return False, "public key binding conflict"
             seen_public_keys[public_key] = node_id
-            sig_payload = build_signature_payload(
-                event_type=event_type,
-                node_id=node_id,
-                sequence=sequence,
-                payload=payload,
-            )
-            if not verify_signature(
-                public_key_b64=public_key,
-                public_key_algo=public_key_algo,
-                signature_hex=signature,
-                payload=sig_payload,
-            ):
+            if event_type == "gate_message":
+                signature_payloads = _private_gate_signature_payload_variants(
+                    str(payload.get("gate", "") or ""),
+                    evt,
+                )
+            else:
+                signature_payloads = [payload]
+            signature_ok = False
+            for signature_payload in signature_payloads:
+                sig_payload = build_signature_payload(
+                    event_type=event_type,
+                    node_id=node_id,
+                    sequence=sequence,
+                    payload=signature_payload,
+                )
+                if verify_signature(
+                    public_key_b64=public_key,
+                    public_key_algo=public_key_algo,
+                    signature_hex=signature,
+                    payload=sig_payload,
+                ):
+                    signature_ok = True
+                    break
+            if not signature_ok:
                 return False, "invalid signature"
             computed = ChainEvent.from_dict(evt).event_id
             if computed != event_id:
                 return False, "event_id mismatch"
-            last_seq[node_id] = sequence
+            last_seq[sequence_key] = sequence
 
         # Apply fork
         self.events = prefix + ordered

backend/services/mesh/mesh_infonet_sync_support.py

@@ -276,5 +276,6 @@ def should_run_sync(
 ) -> bool:
     current_time = int(now if now is not None else time.time())
     if state.last_outcome == "running":
-        return False
+        started_at = int(state.last_sync_started_at or 0)
+        return started_at <= 0 or current_time - started_at >= 300
     return int(state.next_sync_due_at or 0) <= current_time

backend/services/mesh/mesh_schema.py

@@ -2,6 +2,9 @@
 
 from __future__ import annotations
 
+import base64
+import binascii
+import math
 from dataclasses import dataclass
 from typing import Any, Callable
 
@@ -33,6 +36,58 @@ def _require_fields(payload: dict[str, Any], fields: tuple[str, ...]) -> tuple[b
     return True, "ok"
 
 
+def _decode_base64ish(value: Any) -> bytes | None:
+    raw = str(value or "").strip()
+    if not raw or any(ch.isspace() for ch in raw):
+        return None
+    padded = raw + ("=" * (-len(raw) % 4))
+    for altchars in (None, b"-_"):
+        try:
+            return base64.b64decode(padded.encode("ascii"), altchars=altchars, validate=True)
+        except (binascii.Error, UnicodeEncodeError, ValueError):
+            continue
+    return None
+
+
+def _byte_entropy(data: bytes) -> float:
+    if not data:
+        return 0.0
+    counts = [0] * 256
+    for byte in data:
+        counts[byte] += 1
+    total = float(len(data))
+    return -sum((count / total) * math.log2(count / total) for count in counts if count)
+
+
+def _validate_sealed_bytes_field(
+    payload: dict[str, Any],
+    field: str,
+    *,
+    min_bytes: int = 8,
+    entropy_floor: float = 2.5,
+) -> tuple[bool, str]:
+    data = _decode_base64ish(payload.get(field, ""))
+    if data is None:
+        return False, f"{field} must be base64-encoded sealed bytes"
+    if len(data) < min_bytes:
+        return False, f"{field} is too short"
+
+    # Short test vectors and compact envelopes can be low entropy; only apply
+    # heuristics once there is enough material to distinguish a sealed blob
+    # from accidental base64-encoded plaintext.
+    if len(data) >= 32:
+        printable = sum(1 for byte in data if 32 <= byte <= 126 or byte in (9, 10, 13))
+        if printable / len(data) > 0.9:
+            try:
+                data.decode("utf-8")
+                return False, f"{field} looks like encoded plaintext"
+            except UnicodeDecodeError:
+                pass
+        if _byte_entropy(data) < entropy_floor:
+            return False, f"{field} entropy is too low for sealed bytes"
+    return True, "ok"
+
+
 def _validate_message(payload: dict[str, Any]) -> tuple[bool, str]:
     ok, reason = _require_fields(
         payload, ("message", "destination", "channel", "priority", "ephemeral")
@@ -331,6 +386,7 @@ ACTIVE_PUBLIC_LEDGER_EVENT_TYPES: frozenset[str] = frozenset(
 LEGACY_PUBLIC_LEDGER_EVENT_TYPES: frozenset[str] = frozenset(
     {
         "gate_message",
+        "dm_message",
     }
 )
 """Event types that exist historically on the public chain and must remain
@@ -425,6 +481,8 @@ def validate_event_payload(event_type: str, payload: dict[str, Any]) -> tuple[bo
 
 
 def validate_public_ledger_payload(event_type: str, payload: dict[str, Any]) -> tuple[bool, str]:
+    if event_type == "gate_message":
+        return validate_private_gate_ledger_payload(payload)
     if event_type not in PUBLIC_LEDGER_EVENT_TYPES and event_type not in _EXTENSION_VALIDATORS:
         return False, f"{event_type} is not allowed on the public ledger"
     forbidden = sorted(
@@ -441,6 +499,92 @@ def validate_public_ledger_payload(event_type: str, payload: dict[str, Any]) ->
     return True, "ok"
 
 
+_PRIVATE_GATE_LEDGER_ALLOWED_FIELDS: frozenset[str] = frozenset(
+    {
+        "gate",
+        "ciphertext",
+        "nonce",
+        "sender_ref",
+        "format",
+        "epoch",
+        "gate_envelope",
+        "envelope_hash",
+        "reply_to",
+        "transport_lock",
+        "signed_context",
+    }
+)
+
+
+def validate_private_gate_ledger_payload(payload: dict[str, Any]) -> tuple[bool, str]:
+    """Validate ciphertext-only gate events for private Infonet replication."""
+    ok, reason = validate_event_payload("gate_message", payload)
+    if not ok:
+        return ok, reason
+    unexpected = sorted(
+        key
+        for key in payload.keys()
+        if str(key or "").strip().lower() not in _PRIVATE_GATE_LEDGER_ALLOWED_FIELDS
+    )
+    if unexpected:
+        return False, f"private gate ledger payload contains unsupported fields: {', '.join(unexpected)}"
+    if "message" in payload or "_local_plaintext" in payload or "_local_reply_to" in payload:
+        return False, "private gate ledger payload must not contain plaintext"
+    transport_lock = str(payload.get("transport_lock", "") or "").strip().lower()
+    if transport_lock and transport_lock not in {"private", "private_strong", "rns", "onion"}:
+        return False, "gate messages require private transport_lock"
+    ok, reason = _validate_sealed_bytes_field(payload, "ciphertext")
+    if not ok:
+        return ok, reason
+    ok, reason = _validate_sealed_bytes_field(payload, "nonce")
+    if not ok:
+        return ok, reason
+    return True, "ok"
+
+
+_PRIVATE_DM_LEDGER_ALLOWED_FIELDS: frozenset[str] = frozenset(
+    {
+        "recipient_id",
+        "delivery_class",
+        "recipient_token",
+        "ciphertext",
+        "msg_id",
+        "timestamp",
+        "format",
+        "session_welcome",
+        "sender_seal",
+        "relay_salt",
+        "transport_lock",
+        "signed_context",
+    }
+)
+
+
+def validate_private_dm_ledger_payload(payload: dict[str, Any]) -> tuple[bool, str]:
+    """Validate ciphertext-only DM dead-drop events for private Infonet replication."""
+    ok, reason = validate_event_payload("dm_message", payload)
+    if not ok:
+        return ok, reason
+    unexpected = sorted(
+        key
+        for key in payload.keys()
+        if str(key or "").strip().lower() not in _PRIVATE_DM_LEDGER_ALLOWED_FIELDS
+    )
+    if unexpected:
+        return False, f"private DM ledger payload contains unsupported fields: {', '.join(unexpected)}"
+    if "message" in payload or "plaintext" in payload or "_local_plaintext" in payload:
+        return False, "private DM ledger payload must not contain plaintext"
+    transport_lock = str(payload.get("transport_lock", "") or "").strip().lower()
+    if transport_lock != "private_strong":
+        return False, "DM hashchain spool requires private_strong transport_lock"
+    if not str(payload.get("ciphertext", "") or "").strip():
+        return False, "ciphertext cannot be empty"
+    ok, reason = _validate_sealed_bytes_field(payload, "ciphertext")
+    if not ok:
+        return ok, reason
+    return True, "ok"
+
+
 def validate_protocol_fields(protocol_version: str, network_id: str) -> tuple[bool, str]:
     if protocol_version != PROTOCOL_VERSION:
         return False, "Unsupported protocol_version"

backend/tests/mesh/test_gate_write_cutover.py

@@ -87,11 +87,28 @@ def _run_gate_release_once(monkeypatch, *, transport_tier="private_strong"):
 def _patch_for_successful_post(monkeypatch, module):
     """Apply standard monkeypatches so a gate_message post succeeds."""
     import main
+    from services.mesh import mesh_hashchain
 
     _setup_gate_outbox(monkeypatch)
     monkeypatch.setattr(main, "_verify_gate_message_signed_write", lambda **kw: (True, "ok", kw.get("reply_to", "")))
     monkeypatch.setattr(main, "_resolve_envelope_policy", lambda _gate_id: "envelope_disabled")
 
+    def _fake_private_gate_append(**kwargs):
+        return {
+            "event_id": f"ledger-ev-{kwargs.get('sequence', 0)}",
+            "event_type": "gate_message",
+            "node_id": kwargs["node_id"],
+            "payload": dict(kwargs["payload"]),
+            "timestamp": kwargs.get("timestamp", 0) or 123.0,
+            "sequence": kwargs["sequence"],
+            "signature": kwargs["signature"],
+            "public_key": kwargs["public_key"],
+            "public_key_algo": kwargs["public_key_algo"],
+            "protocol_version": kwargs.get("protocol_version", "infonet/2"),
+        }
+
+    monkeypatch.setattr(mesh_hashchain.infonet, "append_private_gate_message", _fake_private_gate_append)
+
     from services.mesh.mesh_reputation import gate_manager, reputation_ledger
 
     monkeypatch.setattr(gate_manager, "can_enter", lambda *a, **kw: (True, "ok"))
@@ -255,19 +272,30 @@ def test_gate_post_preserves_gate_envelope_in_store(monkeypatch):
 
 
 def test_gate_post_advances_sequence(monkeypatch):
-    """validate_and_set_sequence must be called to advance the counter."""
+    """append_private_gate_message must receive the gate sequence."""
     import main
     from services.mesh import mesh_hashchain
 
     _patch_for_successful_post(monkeypatch, main)
 
-    seq_calls = []
+    append_calls = []
 
-    def track_seq(node_id, seq, *, domain=""):
-        seq_calls.append((node_id, seq, domain))
-        return (True, "ok")
+    def track_private_append(**kwargs):
+        append_calls.append(kwargs)
+        return {
+            "event_id": "ev-seq",
+            "event_type": "gate_message",
+            "node_id": kwargs["node_id"],
+            "payload": dict(kwargs["payload"]),
+            "timestamp": kwargs.get("timestamp", 0) or 123.0,
+            "sequence": kwargs["sequence"],
+            "signature": kwargs["signature"],
+            "public_key": kwargs["public_key"],
+            "public_key_algo": kwargs["public_key_algo"],
+            "protocol_version": kwargs.get("protocol_version", "infonet/2"),
+        }
 
-    monkeypatch.setattr(mesh_hashchain.infonet, "validate_and_set_sequence", track_seq)
+    monkeypatch.setattr(mesh_hashchain.infonet, "append_private_gate_message", track_private_append)
     monkeypatch.setattr(
         mesh_hashchain.gate_store,
         "append",
@@ -280,8 +308,9 @@ def test_gate_post_advances_sequence(monkeypatch):
 
     assert result["ok"] is True
     assert result["queued"] is True
-    assert len(seq_calls) == 1
-    assert seq_calls[0] == ("!sb_test1234567890", 42, "gate_message")
+    assert len(append_calls) == 1
+    assert append_calls[0]["node_id"] == "!sb_test1234567890"
+    assert append_calls[0]["sequence"] == 42
 
 
 def test_gate_post_rejects_replay_via_sequence(monkeypatch):
@@ -290,11 +319,11 @@ def test_gate_post_rejects_replay_via_sequence(monkeypatch):
     from services.mesh import mesh_hashchain
 
     _patch_for_successful_post(monkeypatch, main)
-    monkeypatch.setattr(
-        mesh_hashchain.infonet,
-        "validate_and_set_sequence",
-        lambda node_id, seq: (False, "Replay detected: sequence 1 <= last 1"),
-    )
+
+    def reject_private_append(**_kwargs):
+        raise ValueError("Replay detected: sequence 1 <= last 1")
+
+    monkeypatch.setattr(mesh_hashchain.infonet, "append_private_gate_message", reject_private_append)
 
     gate_id = "infonet"
     body = _build_gate_message_body(gate_id, sequence=1)

backend/tests/mesh/test_mesh_infonet_sync_support.py

@@ -117,3 +117,11 @@ def test_finish_solo_sync_marks_first_node_ready_without_peer_failure():
     assert finished.next_sync_due_at == 500
     assert should_run_sync(finished, now=499) is False
     assert should_run_sync(finished, now=500) is True
+
+
+def test_should_run_sync_recovers_stale_running_state():
+    fresh = SyncWorkerState(last_sync_started_at=100, last_outcome="running")
+    stale = SyncWorkerState(last_sync_started_at=100, last_outcome="running")
+
+    assert should_run_sync(fresh, now=399) is False
+    assert should_run_sync(stale, now=400) is True

backend/tests/mesh/test_mesh_node_bootstrap_runtime.py

@@ -8,6 +8,53 @@ from cryptography.hazmat.primitives.asymmetric import ed25519
 from httpx import ASGITransport, AsyncClient
 
 
+def test_onion_peer_requests_use_arti_socks_proxy(monkeypatch):
+    import main
+    from services import wormhole_supervisor
+
+    monkeypatch.setattr(main, "_infonet_private_transport_required", lambda: True)
+    monkeypatch.setattr(
+        main,
+        "get_settings",
+        lambda: SimpleNamespace(MESH_ARTI_ENABLED=True, MESH_ARTI_SOCKS_PORT=19050),
+    )
+    monkeypatch.setattr(wormhole_supervisor, "_check_arti_ready", lambda: True)
+
+    proxies = main._infonet_peer_requests_proxies("http://exampleabcd.onion:8000")
+
+    assert proxies == {
+        "http": "socks5h://127.0.0.1:19050",
+        "https": "socks5h://127.0.0.1:19050",
+    }
+
+
+def test_private_peer_requests_reject_clearnet(monkeypatch):
+    import main
+
+    monkeypatch.setattr(main, "_infonet_private_transport_required", lambda: True)
+
+    try:
+        main._infonet_peer_requests_proxies("https://seed.example")
+    except RuntimeError as exc:
+        assert "private Infonet requires onion/RNS transport" in str(exc)
+    else:
+        raise AssertionError("clearnet peer was allowed while private transport is required")
+
+
+def test_local_peer_url_prefers_configured_public_peer_url(monkeypatch):
+    import main
+
+    monkeypatch.setattr(
+        main,
+        "get_settings",
+        lambda: SimpleNamespace(
+            MESH_PUBLIC_PEER_URL="HTTP://LOCALPEEREXAMPLE.onion:8000/",
+        ),
+    )
+
+    assert main._local_infonet_peer_url() == "http://localpeerexample.onion:8000"
+
+
 def _write_signed_manifest(path, *, private_key):
     from services.mesh.mesh_bootstrap_manifest import BOOTSTRAP_MANIFEST_VERSION
     from services.mesh.mesh_crypto import canonical_json
@@ -142,6 +189,134 @@ def test_refresh_node_peer_store_suppresses_clearnet_seed_by_default(tmp_path, m
     assert store.records_for_bucket("sync") == []
 
 
+def test_refresh_node_peer_store_prunes_persisted_clearnet_records_in_private_mode(tmp_path, monkeypatch):
+    import main
+    from services.config import get_settings
+    from services.mesh import mesh_peer_store as peer_store_mod
+
+    peer_store_path = tmp_path / "peer_store.json"
+    monkeypatch.setattr(peer_store_mod, "DEFAULT_PEER_STORE_PATH", peer_store_path)
+    store = peer_store_mod.PeerStore(peer_store_path)
+    store.upsert(
+        peer_store_mod.make_bootstrap_peer_record(
+            peer_url="https://node.shadowbroker.info",
+            transport="clearnet",
+            role="seed",
+            signer_id="shadowbroker-default",
+            now=1_749_999_900,
+        )
+    )
+    store.upsert(
+        peer_store_mod.make_sync_peer_record(
+            peer_url="https://node.shadowbroker.info",
+            transport="clearnet",
+            role="seed",
+            source="bundle",
+            now=1_749_999_900,
+        )
+    )
+    store.upsert(
+        peer_store_mod.make_push_peer_record(
+            peer_url="https://node.shadowbroker.info",
+            transport="clearnet",
+            role="relay",
+            now=1_749_999_900,
+        )
+    )
+    store.save()
+
+    onion_seed = "http://gqpbunqbgtkcqilvclm3xrkt3zowjyl3s62kkktvojgvxzizamvbrqid.onion:8000"
+    monkeypatch.setenv("MESH_RELAY_PEERS", "")
+    monkeypatch.setenv("MESH_BOOTSTRAP_SEED_PEERS", onion_seed)
+    monkeypatch.setenv("MESH_DEFAULT_SYNC_PEERS", "")
+    monkeypatch.delenv("MESH_INFONET_ALLOW_CLEARNET_SYNC", raising=False)
+    monkeypatch.setenv("MESH_BOOTSTRAP_SIGNER_PUBLIC_KEY", "")
+    get_settings.cache_clear()
+
+    try:
+        snapshot = main._refresh_node_peer_store(now=1_750_000_000)
+        store = peer_store_mod.PeerStore(peer_store_path)
+        store.load()
+    finally:
+        get_settings.cache_clear()
+
+    assert snapshot["private_transport_required"] is True
+    assert snapshot["pruned_clearnet_peer_count"] == 3
+    assert [record.peer_url for record in store.records()] == [onion_seed, onion_seed]
+    assert {record.bucket for record in store.records()} == {"bootstrap", "sync"}
+    assert all(record.transport == "onion" for record in store.records())
+
+
+def test_infonet_peer_url_filter_excludes_clearnet_in_private_mode(monkeypatch):
+    import main
+    from services.config import get_settings
+
+    monkeypatch.delenv("MESH_INFONET_ALLOW_CLEARNET_SYNC", raising=False)
+    get_settings.cache_clear()
+
+    try:
+        assert main._filter_infonet_peer_urls(
+            [
+                "https://node.shadowbroker.info",
+                "http://gqpbunqbgtkcqilvclm3xrkt3zowjyl3s62kkktvojgvxzizamvbrqid.onion:8000",
+            ]
+        ) == ["http://gqpbunqbgtkcqilvclm3xrkt3zowjyl3s62kkktvojgvxzizamvbrqid.onion:8000"]
+    finally:
+        get_settings.cache_clear()
+
+
+def test_public_sync_cycle_backs_off_on_429_retry_after(tmp_path, monkeypatch):
+    import time
+
+    import main
+    from services.config import get_settings
+    from services.mesh import mesh_peer_store as peer_store_mod
+
+    peer_store_path = tmp_path / "peer_store.json"
+    monkeypatch.setattr(peer_store_mod, "DEFAULT_PEER_STORE_PATH", peer_store_path)
+    onion_seed = "http://gqpbunqbgtkcqilvclm3xrkt3zowjyl3s62kkktvojgvxzizamvbrqid.onion:8000"
+    store = peer_store_mod.PeerStore(peer_store_path)
+    store.upsert(
+        peer_store_mod.make_sync_peer_record(
+            peer_url=onion_seed,
+            transport="onion",
+            role="seed",
+            source="bundle",
+            now=1_750_000_000,
+        )
+    )
+    store.save()
+
+    monkeypatch.delenv("MESH_INFONET_ALLOW_CLEARNET_SYNC", raising=False)
+    monkeypatch.setenv("MESH_SYNC_FAILURE_BACKOFF_S", "60")
+    monkeypatch.setenv("MESH_BOOTSTRAP_SEED_FAILURE_COOLDOWN_S", "15")
+    get_settings.cache_clear()
+    monkeypatch.setattr(main, "_participant_node_enabled", lambda: True)
+    monkeypatch.setattr(main, "_ensure_infonet_private_transport_ready", lambda reason="": True)
+    monkeypatch.setattr(
+        main,
+        "_sync_from_peer",
+        lambda peer_url: (_ for _ in ()).throw(
+            main.PeerSyncHTTPError(429, "rate limited", retry_after_s=180)
+        ),
+    )
+    main.set_sync_state(main.SyncWorkerState())
+
+    try:
+        before = int(time.time())
+        state = main._run_public_sync_cycle()
+        store = peer_store_mod.PeerStore(peer_store_path)
+        store.load()
+    finally:
+        get_settings.cache_clear()
+        main.set_sync_state(main.SyncWorkerState())
+
+    record = store.records_for_bucket("sync")[0]
+    assert state.last_error == "HTTP 429: rate limited"
+    assert state.next_sync_due_at >= before + 180
+    assert record.cooldown_until >= before + 180
+
+
 def test_verify_peer_push_hmac_requires_allowlisted_peer(monkeypatch):
     import hashlib
     import hmac
@@ -225,3 +400,29 @@ def test_public_sync_cycle_allows_first_node_without_peers(tmp_path, monkeypatch
     assert result.last_error == ""
     assert result.last_peer_url == ""
     assert result.consecutive_failures == 0
+
+
+def test_headless_mesh_node_runtime_is_explicit(monkeypatch):
+    import main
+
+    monkeypatch.setattr(main, "_MESH_ONLY", True)
+    monkeypatch.setattr(main, "_HEADLESS_MESH_NODE_RUNTIME", False)
+    assert main._infonet_node_runtime_requested() is False
+
+    monkeypatch.setattr(main, "_HEADLESS_MESH_NODE_RUNTIME", True)
+    assert main._infonet_node_runtime_requested() is True
+
+
+def test_meshnode_scripts_enable_private_hashchain_runtime():
+    from pathlib import Path
+
+    root = Path(__file__).resolve().parents[3]
+    bat = (root / "meshnode.bat").read_text(encoding="utf-8")
+    sh = (root / "meshnode.sh").read_text(encoding="utf-8")
+
+    for script in (bat, sh):
+        assert "SHADOWBROKER_MESH_NODE_RUNTIME=true" in script
+        assert "MESH_INFONET_ALLOW_CLEARNET_SYNC=false" in script
+        assert "MESH_ARTI_ENABLED=true" in script
+        assert "MESH_DM_HASHCHAIN_SPOOL_LIMIT=2" in script
+        assert "gqpbunqbgtkcqilvclm3xrkt3zowjyl3s62kkktvojgvxzizamvbrqid.onion:8000" in script

backend/tests/mesh/test_private_dm_hashchain.py

@@ -0,0 +1,213 @@
+import base64
+import time
+
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.asymmetric import ed25519
+
+from services.config import get_settings
+from services.mesh import mesh_crypto, mesh_dm_relay, mesh_hashchain, mesh_protocol, mesh_secure_storage
+
+
+def _keypair():
+    private_key = ed25519.Ed25519PrivateKey.generate()
+    public_raw = private_key.public_key().public_bytes(
+        encoding=serialization.Encoding.Raw,
+        format=serialization.PublicFormat.Raw,
+    )
+    public_key = base64.b64encode(public_raw).decode("utf-8")
+    node_id = mesh_crypto.derive_node_id(public_key)
+    return private_key, public_key, node_id
+
+
+def _payload(recipient_id: str = "recipient-a", msg_id: str = "dm-1") -> dict:
+    return mesh_protocol.normalize_payload(
+        "dm_message",
+        {
+            "recipient_id": recipient_id,
+            "delivery_class": "request",
+            "recipient_token": "",
+            "ciphertext": base64.b64encode(f"cipher-{msg_id}".encode("utf-8")).decode("ascii"),
+            "msg_id": msg_id,
+            "timestamp": int(time.time()),
+            "format": "mls1",
+            "transport_lock": "private_strong",
+        },
+    )
+
+
+def _signature(private_key, node_id: str, sequence: int, payload: dict) -> str:
+    signature_payload = mesh_crypto.build_signature_payload(
+        event_type="dm_message",
+        node_id=node_id,
+        sequence=sequence,
+        payload=payload,
+    )
+    return private_key.sign(signature_payload.encode("utf-8")).hex()
+
+
+def _fresh_infonet(tmp_path, monkeypatch) -> mesh_hashchain.Infonet:
+    monkeypatch.setattr(mesh_hashchain, "DATA_DIR", tmp_path)
+    monkeypatch.setattr(mesh_hashchain, "CHAIN_FILE", tmp_path / "infonet.json")
+    monkeypatch.setattr(mesh_hashchain, "WAL_FILE", tmp_path / "infonet.wal")
+    return mesh_hashchain.Infonet()
+
+
+def _fresh_relay(tmp_path, monkeypatch) -> mesh_dm_relay.DMRelay:
+    monkeypatch.setattr(mesh_dm_relay, "DATA_DIR", tmp_path)
+    monkeypatch.setattr(mesh_dm_relay, "RELAY_FILE", tmp_path / "dm_relay.json")
+    monkeypatch.setattr(mesh_secure_storage, "DATA_DIR", tmp_path)
+    monkeypatch.setattr(mesh_secure_storage, "MASTER_KEY_FILE", tmp_path / "wormhole_secure_store.key")
+    get_settings.cache_clear()
+    return mesh_dm_relay.DMRelay()
+
+
+def test_private_dm_hashchain_spools_two_ciphertexts_per_recipient_from_distinct_senders(tmp_path, monkeypatch):
+    inf = _fresh_infonet(tmp_path, monkeypatch)
+    senders = [_keypair(), _keypair()]
+
+    for idx, (private_key, public_key, node_id) in enumerate(senders, start=1):
+        payload = _payload(msg_id=f"dm-{idx}")
+        event = inf.append_private_dm_message(
+            node_id=node_id,
+            payload=payload,
+            signature=_signature(private_key, node_id, 1, payload),
+            sequence=1,
+            public_key=public_key,
+            public_key_algo="Ed25519",
+            protocol_version=mesh_protocol.PROTOCOL_VERSION,
+            timestamp=float(payload["timestamp"]),
+        )
+        assert event["event_type"] == "dm_message"
+
+    private_key, public_key, node_id = _keypair()
+    third = _payload(msg_id="dm-3")
+    try:
+        inf.append_private_dm_message(
+            node_id=node_id,
+            payload=third,
+            signature=_signature(private_key, node_id, 1, third),
+            sequence=1,
+            public_key=public_key,
+            public_key_algo="Ed25519",
+            protocol_version=mesh_protocol.PROTOCOL_VERSION,
+            timestamp=float(third["timestamp"]),
+        )
+    except ValueError as exc:
+        assert "spool full" in str(exc)
+    else:
+        raise AssertionError("third DM spool event was accepted")
+
+    for _private_key, _public_key, sender_node_id in senders:
+        assert inf.sequence_domains[f"{sender_node_id}|dm_message"] == 1
+    assert inf.validate_chain(verify_signatures=True)[0] is True
+
+
+def test_private_dm_hashchain_limits_one_active_spool_per_sender_recipient_pair(tmp_path, monkeypatch):
+    inf = _fresh_infonet(tmp_path, monkeypatch)
+    private_key, public_key, node_id = _keypair()
+
+    first = _payload(msg_id="dm-1")
+    inf.append_private_dm_message(
+        node_id=node_id,
+        payload=first,
+        signature=_signature(private_key, node_id, 1, first),
+        sequence=1,
+        public_key=public_key,
+        public_key_algo="Ed25519",
+        protocol_version=mesh_protocol.PROTOCOL_VERSION,
+        timestamp=float(first["timestamp"]),
+    )
+
+    second = _payload(msg_id="dm-2")
+    try:
+        inf.append_private_dm_message(
+            node_id=node_id,
+            payload=second,
+            signature=_signature(private_key, node_id, 2, second),
+            sequence=2,
+            public_key=public_key,
+            public_key_algo="Ed25519",
+            protocol_version=mesh_protocol.PROTOCOL_VERSION,
+            timestamp=float(second["timestamp"]),
+        )
+    except ValueError as exc:
+        assert "sender spool full" in str(exc)
+    else:
+        raise AssertionError("second DM from same sender to same recipient was accepted")
+
+
+def test_private_dm_hashchain_rejects_plaintext(tmp_path, monkeypatch):
+    inf = _fresh_infonet(tmp_path, monkeypatch)
+    private_key, public_key, node_id = _keypair()
+    payload = _payload()
+    payload["message"] = "plaintext"
+
+    try:
+        inf.append_private_dm_message(
+            node_id=node_id,
+            payload=payload,
+            signature=_signature(private_key, node_id, 1, _payload()),
+            sequence=1,
+            public_key=public_key,
+            public_key_algo="Ed25519",
+            protocol_version=mesh_protocol.PROTOCOL_VERSION,
+        )
+    except ValueError as exc:
+        assert "plaintext" in str(exc)
+    else:
+        raise AssertionError("private DM append accepted plaintext")
+
+
+def test_private_dm_hashchain_rejects_non_sealed_ciphertext_shape(tmp_path, monkeypatch):
+    inf = _fresh_infonet(tmp_path, monkeypatch)
+    private_key, public_key, node_id = _keypair()
+    payload = _payload()
+    payload["ciphertext"] = "not sealed plaintext"
+
+    try:
+        inf.append_private_dm_message(
+            node_id=node_id,
+            payload=payload,
+            signature=_signature(private_key, node_id, 1, payload),
+            sequence=1,
+            public_key=public_key,
+            public_key_algo="Ed25519",
+            protocol_version=mesh_protocol.PROTOCOL_VERSION,
+        )
+    except ValueError as exc:
+        assert "sealed bytes" in str(exc)
+    else:
+        raise AssertionError("private DM append accepted non-base64 ciphertext")
+
+
+def test_hydrate_dm_relay_from_chain_delivers_to_poll_claim(tmp_path, monkeypatch):
+    inf = _fresh_infonet(tmp_path / "chain", monkeypatch)
+    relay = _fresh_relay(tmp_path / "relay", monkeypatch)
+    monkeypatch.setattr(mesh_hashchain, "infonet", inf)
+    monkeypatch.setattr(mesh_dm_relay, "dm_relay", relay)
+
+    private_key, public_key, node_id = _keypair()
+    payload = _payload(recipient_id="recipient-a", msg_id="dm-chain-1")
+    event = inf.append_private_dm_message(
+        node_id=node_id,
+        payload=payload,
+        signature=_signature(private_key, node_id, 1, payload),
+        sequence=1,
+        public_key=public_key,
+        public_key_algo="Ed25519",
+        protocol_version=mesh_protocol.PROTOCOL_VERSION,
+        timestamp=float(payload["timestamp"]),
+    )
+
+    from main import _hydrate_dm_relay_from_chain
+
+    assert _hydrate_dm_relay_from_chain([event]) == 1
+    messages, more = relay.collect_claims(
+        "recipient-a",
+        [{"type": "requests", "token": "recipient-request-token"}],
+        limit=8,
+    )
+
+    assert more is False
+    assert [message["msg_id"] for message in messages] == ["dm-chain-1"]
+    assert messages[0]["ciphertext"] == payload["ciphertext"]

backend/tests/mesh/test_private_gate_hashchain.py

@@ -0,0 +1,269 @@
+import base64
+
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.asymmetric import ed25519
+
+from services.mesh import mesh_crypto, mesh_hashchain, mesh_protocol
+
+
+def _keypair():
+    private_key = ed25519.Ed25519PrivateKey.generate()
+    public_raw = private_key.public_key().public_bytes(
+        encoding=serialization.Encoding.Raw,
+        format=serialization.PublicFormat.Raw,
+    )
+    public_key = base64.b64encode(public_raw).decode("utf-8")
+    node_id = mesh_crypto.derive_node_id(public_key)
+    return private_key, public_key, node_id
+
+
+def _sign(private_key, *, event_type: str, node_id: str, sequence: int, payload: dict) -> str:
+    signature_payload = mesh_crypto.build_signature_payload(
+        event_type=event_type,
+        node_id=node_id,
+        sequence=sequence,
+        payload=payload,
+    )
+    return private_key.sign(signature_payload.encode("utf-8")).hex()
+
+
+def _message_payload(text: str) -> dict:
+    return mesh_protocol.normalize_payload(
+        "message",
+        {
+            "message": text,
+            "destination": "broadcast",
+            "channel": "LongFast",
+            "priority": "normal",
+            "ephemeral": False,
+        },
+    )
+
+
+def _gate_payload(gate_id: str = "ops-gate", *, epoch: int = 2, plaintext: bool = False) -> dict:
+    payload = {
+        "gate": gate_id,
+        "ciphertext": base64.b64encode(b"encrypted-gate-ciphertext").decode("ascii"),
+        "nonce": base64.b64encode(b"nonce-value-1234").decode("ascii"),
+        "sender_ref": "sender-ref-1",
+        "format": "mls1",
+        "transport_lock": "private_strong",
+    }
+    if epoch > 0:
+        payload["epoch"] = epoch
+    if plaintext:
+        payload["message"] = "this must never land on the chain"
+    return mesh_protocol.normalize_payload("gate_message", payload) if not plaintext else payload
+
+
+def _gate_event(
+    private_key,
+    public_key: str,
+    node_id: str,
+    *,
+    sequence: int,
+    prev_hash: str,
+    payload: dict,
+    signature_payload: dict | None = None,
+) -> dict:
+    signature = _sign(
+        private_key,
+        event_type="gate_message",
+        node_id=node_id,
+        sequence=sequence,
+        payload=signature_payload or payload,
+    )
+    return mesh_hashchain.ChainEvent(
+        prev_hash=prev_hash,
+        event_type="gate_message",
+        node_id=node_id,
+        payload=payload,
+        timestamp=1234.0 + sequence,
+        sequence=sequence,
+        signature=signature,
+        public_key=public_key,
+        public_key_algo="Ed25519",
+        protocol_version=mesh_protocol.PROTOCOL_VERSION,
+        network_id=mesh_protocol.NETWORK_ID,
+    ).to_dict()
+
+
+def _fresh_infonet(tmp_path, monkeypatch) -> mesh_hashchain.Infonet:
+    monkeypatch.setattr(mesh_hashchain, "DATA_DIR", tmp_path)
+    monkeypatch.setattr(mesh_hashchain, "CHAIN_FILE", tmp_path / "infonet.json")
+    monkeypatch.setattr(mesh_hashchain, "WAL_FILE", tmp_path / "infonet.wal")
+    return mesh_hashchain.Infonet()
+
+
+def test_private_gate_fork_uses_gate_sequence_domain_and_signature_variants(tmp_path, monkeypatch):
+    inf = _fresh_infonet(tmp_path, monkeypatch)
+    private_key, public_key, node_id = _keypair()
+
+    public_payload = _message_payload("public prefix")
+    public_event = inf.append(
+        event_type="message",
+        node_id=node_id,
+        payload=public_payload,
+        sequence=1,
+        signature=_sign(
+            private_key,
+            event_type="message",
+            node_id=node_id,
+            sequence=1,
+            payload=public_payload,
+        ),
+        public_key=public_key,
+        public_key_algo="Ed25519",
+        protocol_version=mesh_protocol.PROTOCOL_VERSION,
+    )
+
+    gate_payload = _gate_payload(epoch=3)
+    signature_payload = dict(gate_payload)
+    signature_payload.pop("epoch", None)
+    gate_event = _gate_event(
+        private_key,
+        public_key,
+        node_id,
+        sequence=1,
+        prev_hash=public_event["event_id"],
+        payload=gate_payload,
+        signature_payload=signature_payload,
+    )
+
+    ok, reason = inf.apply_fork([gate_event], gate_event["event_id"], proof_count=2, quorum=2)
+
+    assert ok is True, reason
+    assert inf.events[-1]["event_type"] == "gate_message"
+    assert inf.node_sequences[node_id] == 1
+    assert inf.sequence_domains[f"{node_id}|gate_message"] == 1
+    assert inf.validate_chain(verify_signatures=True)[0] is True
+
+
+def test_private_gate_fork_rejects_plaintext_payload(tmp_path, monkeypatch):
+    inf = _fresh_infonet(tmp_path, monkeypatch)
+    private_key, public_key, node_id = _keypair()
+
+    public_payload = _message_payload("public prefix")
+    public_event = inf.append(
+        event_type="message",
+        node_id=node_id,
+        payload=public_payload,
+        sequence=1,
+        signature=_sign(
+            private_key,
+            event_type="message",
+            node_id=node_id,
+            sequence=1,
+            payload=public_payload,
+        ),
+        public_key=public_key,
+        public_key_algo="Ed25519",
+        protocol_version=mesh_protocol.PROTOCOL_VERSION,
+    )
+
+    plaintext_payload = _gate_payload(plaintext=True)
+    gate_event = _gate_event(
+        private_key,
+        public_key,
+        node_id,
+        sequence=1,
+        prev_hash=public_event["event_id"],
+        payload=plaintext_payload,
+    )
+
+    ok, reason = inf.apply_fork([gate_event], gate_event["event_id"], proof_count=2, quorum=2)
+
+    assert ok is False
+    assert "normalized" in reason or "plaintext" in reason
+    assert len(inf.events) == 1
+    assert "gate_message" not in inf.get_info()["event_types"]
+
+
+def test_append_private_gate_message_rejects_plaintext_before_normalizing(tmp_path, monkeypatch):
+    inf = _fresh_infonet(tmp_path, monkeypatch)
+    private_key, public_key, node_id = _keypair()
+    payload = _gate_payload()
+    payload["message"] = "plaintext should not be silently dropped"
+
+    try:
+        inf.append_private_gate_message(
+            node_id=node_id,
+            payload=payload,
+            sequence=1,
+            signature=_sign(
+                private_key,
+                event_type="gate_message",
+                node_id=node_id,
+                sequence=1,
+                payload=_gate_payload(),
+            ),
+            public_key=public_key,
+            public_key_algo="Ed25519",
+            protocol_version=mesh_protocol.PROTOCOL_VERSION,
+        )
+    except ValueError as exc:
+        assert "plaintext" in str(exc)
+    else:
+        raise AssertionError("private gate append accepted plaintext")
+
+    assert inf.events == []
+
+
+def test_append_private_gate_message_requires_private_strong_transport_lock(tmp_path, monkeypatch):
+    inf = _fresh_infonet(tmp_path, monkeypatch)
+    private_key, public_key, node_id = _keypair()
+    payload = _gate_payload()
+    payload.pop("transport_lock", None)
+
+    try:
+        inf.append_private_gate_message(
+            node_id=node_id,
+            payload=payload,
+            sequence=1,
+            signature=_sign(
+                private_key,
+                event_type="gate_message",
+                node_id=node_id,
+                sequence=1,
+                payload=_gate_payload(),
+            ),
+            public_key=public_key,
+            public_key_algo="Ed25519",
+            protocol_version=mesh_protocol.PROTOCOL_VERSION,
+        )
+    except ValueError as exc:
+        assert "private_strong" in str(exc)
+    else:
+        raise AssertionError("private gate append accepted missing transport_lock")
+
+    assert inf.events == []
+
+
+def test_append_private_gate_message_rejects_non_sealed_ciphertext_shape(tmp_path, monkeypatch):
+    inf = _fresh_infonet(tmp_path, monkeypatch)
+    private_key, public_key, node_id = _keypair()
+    payload = _gate_payload()
+    payload["ciphertext"] = "not sealed plaintext"
+
+    try:
+        inf.append_private_gate_message(
+            node_id=node_id,
+            payload=payload,
+            sequence=1,
+            signature=_sign(
+                private_key,
+                event_type="gate_message",
+                node_id=node_id,
+                sequence=1,
+                payload=payload,
+            ),
+            public_key=public_key,
+            public_key_algo="Ed25519",
+            protocol_version=mesh_protocol.PROTOCOL_VERSION,
+        )
+    except ValueError as exc:
+        assert "sealed bytes" in str(exc)
+    else:
+        raise AssertionError("private gate append accepted non-base64 ciphertext")
+
+    assert inf.events == []

backend/tests/mesh/test_s14b_public_sync_gate_filter.py

@@ -1,14 +1,12 @@
-"""S14B Public Sync Gate Event Filter.
+"""S14B private sync gate event policy.
 
-Tests:
-- GET /api/mesh/infonet/sync excludes gate_message when local infonet contains legacy gate_message plus public events
-- POST /api/mesh/infonet/sync excludes gate_message under the same condition
-- Both main app and router-served paths are covered
-- Non-gate public redactions still hold (vote gate label stripped, key_rotate identity stripped)
-- Do not overclaim that gate_message is removed from historical infonet storage or ingest
+Private Infonet sync carries encrypted gate_message ledger events. If a node
+is configured to allow clearnet-compatible sync, those gate events are filtered
+out of the sync response.
 """
 
 import asyncio
+import base64
 import json
 
 from starlette.requests import Request
@@ -17,9 +15,6 @@ import main
 from services.mesh import mesh_hashchain
 
 
-# ── Helpers ──────────────────────────────────────────────────────────────
-
-
 def _message_event() -> dict:
     return {
         "event_id": "msg-1",
@@ -83,6 +78,7 @@ def _gate_message_event() -> dict:
             "nonce": "nonce-1",
             "sender_ref": "sender-ref-1",
             "format": "mls1",
+            "transport_lock": "private_strong",
         },
         "timestamp": 103.0,
         "sequence": 4,
@@ -93,9 +89,31 @@ def _gate_message_event() -> dict:
     }
 
 
-class _FakeInfonet:
-    """Minimal fake infonet with a gate_message among public events."""
+def _dm_message_event() -> dict:
+    return {
+        "event_id": "dm-1",
+        "event_type": "dm_message",
+        "node_id": "!node-5",
+        "payload": {
+            "recipient_id": "recipient-a",
+            "delivery_class": "request",
+            "recipient_token": "",
+            "ciphertext": base64.b64encode(b"sealed-dm-ciphertext").decode("ascii"),
+            "msg_id": "dm-1",
+            "timestamp": 104,
+            "format": "mls1",
+            "transport_lock": "private_strong",
+        },
+        "timestamp": 104.0,
+        "sequence": 5,
+        "signature": "sig",
+        "public_key": "pub",
+        "public_key_algo": "Ed25519",
+        "protocol_version": "infonet/2",
+    }
 
+
+class _FakeInfonet:
     def __init__(self):
         self.head_hash = "head-1"
         self.events = [
@@ -113,12 +131,10 @@ class _FakeInfonet:
             return int(getattr(limit, "default", 100) or 100)
 
     def get_events_after(self, after_hash: str, limit=100):
-        resolved = self._limit_value(limit)
-        return [dict(e) for e in self.events[:resolved]]
+        return [dict(e) for e in self.events[: self._limit_value(limit)]]
 
     def get_events_after_locator(self, locator: list[str], limit=100):
-        resolved = self._limit_value(limit)
-        return self.head_hash, 0, [dict(e) for e in self.events[:resolved]]
+        return self.head_hash, 0, [dict(e) for e in self.events[: self._limit_value(limit)]]
 
     def get_merkle_proofs(self, start_index: int, count: int):
         return {"root": "merkle-root", "total": len(self.events), "start": start_index, "proofs": []}
@@ -127,7 +143,7 @@ class _FakeInfonet:
         return "merkle-root"
 
 
-def _json_request(path: str, body: dict) -> Request:
+def _json_request(path: str, body: dict, *, client_host: str = "127.0.0.1", headers: dict[str, str] | None = None) -> Request:
     payload = json.dumps(body).encode("utf-8")
     sent = {"value": False}
 
@@ -137,11 +153,14 @@ def _json_request(path: str, body: dict) -> Request:
         sent["value"] = True
         return {"type": "http.request", "body": payload, "more_body": False}
 
+    raw_headers = [(b"content-type", b"application/json")]
+    for key, value in dict(headers or {}).items():
+        raw_headers.append((key.lower().encode("ascii"), str(value).encode("ascii")))
     return Request(
         {
             "type": "http",
-            "headers": [(b"content-type", b"application/json")],
-            "client": ("test", 12345),
+            "headers": raw_headers,
+            "client": (client_host, 12345),
             "method": "POST",
             "path": path,
         },
@@ -149,20 +168,15 @@ def _json_request(path: str, body: dict) -> Request:
     )
 
 
-def _get_request(path: str) -> Request:
-    sent = {"value": False}
-
+def _get_request(path: str, *, client_host: str = "127.0.0.1", headers: dict[str, str] | None = None) -> Request:
     async def receive():
-        if sent["value"]:
-            return {"type": "http.request", "body": b"", "more_body": False}
-        sent["value"] = True
         return {"type": "http.request", "body": b"", "more_body": False}
 
     return Request(
         {
             "type": "http",
-            "headers": [],
-            "client": ("test", 12345),
+            "headers": [(key.lower().encode("ascii"), str(value).encode("ascii")) for key, value in dict(headers or {}).items()],
+            "client": (client_host, 12345),
             "method": "GET",
             "path": path,
         },
@@ -170,120 +184,166 @@ def _get_request(path: str) -> Request:
     )
 
 
-# ── GET sync excludes gate_message (main app) ──────────────────────────
+def _force_private_sync(monkeypatch):
+    monkeypatch.setattr(main, "_infonet_private_transport_required", lambda: True)
+    monkeypatch.setattr(main, "_request_appears_private_infonet_transport", lambda request: True)
 
 
-def test_get_sync_excludes_gate_message(client, monkeypatch):
-    """GET /api/mesh/infonet/sync must not return gate_message events."""
+def _force_private_policy_only(monkeypatch):
+    monkeypatch.setattr(main, "_infonet_private_transport_required", lambda: True)
+
+
+def _force_clearnet_sync(monkeypatch):
+    monkeypatch.setattr(main, "_infonet_private_transport_required", lambda: False)
+
+
+def _event_types(events: list[dict]) -> list[str]:
+    return [str(e.get("event_type", "")) for e in events]
+
+
+def test_private_sync_redacts_private_events_from_exposed_clearnet_request(monkeypatch):
+    _force_private_policy_only(monkeypatch)
+    request = _get_request("/api/mesh/infonet/sync", client_host="203.0.113.10")
+
+    events = main._infonet_sync_response_events(
+        [_message_event(), _gate_message_event(), _dm_message_event()],
+        request=request,
+    )
+
+    assert _event_types(events) == ["message"]
+
+
+def test_private_sync_includes_private_events_for_loopback_request(monkeypatch):
+    _force_private_policy_only(monkeypatch)
+    request = _get_request("/api/mesh/infonet/sync", client_host="127.0.0.1")
+
+    events = main._infonet_sync_response_events(
+        [_message_event(), _gate_message_event(), _dm_message_event()],
+        request=request,
+    )
+
+    assert _event_types(events) == ["message", "gate_message", "dm_message"]
+
+
+def test_private_sync_redacts_private_events_when_forwarded_for_is_clearnet(monkeypatch):
+    _force_private_policy_only(monkeypatch)
+    request = _get_request(
+        "/api/mesh/infonet/sync",
+        client_host="127.0.0.1",
+        headers={"x-forwarded-for": "198.51.100.44"},
+    )
+
+    events = main._infonet_sync_response_events(
+        [_message_event(), _gate_message_event(), _dm_message_event()],
+        request=request,
+    )
+
+    assert _event_types(events) == ["message"]
+
+
+def test_get_sync_includes_gate_message_on_private_transport(client, monkeypatch):
+    _force_private_sync(monkeypatch)
     monkeypatch.setattr(mesh_hashchain, "infonet", _FakeInfonet(), raising=False)
-    resp = client.get("/api/mesh/infonet/sync")
-    data = resp.json()
-    event_types = [e["event_type"] for e in data["events"]]
-    assert "gate_message" not in event_types
-    assert "message" in event_types
-    assert "vote" in event_types
-    assert "key_rotate" in event_types
+
+    data = client.get("/api/mesh/infonet/sync").json()
+
+    assert "gate_message" in _event_types(data["events"])
+    assert data["count"] == 4
 
 
-def test_get_sync_count_excludes_gate_message(client, monkeypatch):
-    """GET sync count field must reflect filtered events (gate_message excluded)."""
+def test_post_sync_includes_gate_message_on_private_transport(monkeypatch):
+    _force_private_sync(monkeypatch)
     monkeypatch.setattr(mesh_hashchain, "infonet", _FakeInfonet(), raising=False)
-    resp = client.get("/api/mesh/infonet/sync")
-    data = resp.json()
-    assert data["count"] == 3  # message, vote, key_rotate — not gate_message
 
-
-# ── POST sync excludes gate_message (main app) ─────────────────────────
-
-
-def test_post_sync_excludes_gate_message(monkeypatch):
-    """POST /api/mesh/infonet/sync must not return gate_message events."""
-    monkeypatch.setattr(mesh_hashchain, "infonet", _FakeInfonet(), raising=False)
     result = asyncio.run(
         main.infonet_sync_post(
             _json_request("/api/mesh/infonet/sync", {"locator": ["head-1"]})
         )
     )
-    event_types = [e["event_type"] for e in result["events"]]
-    assert "gate_message" not in event_types
-    assert "message" in event_types
-    assert "vote" in event_types
-    assert "key_rotate" in event_types
+
+    assert "gate_message" in _event_types(result["events"])
+    assert result["count"] == 4
 
 
-def test_post_sync_count_excludes_gate_message(monkeypatch):
-    """POST sync count field must reflect filtered events."""
-    monkeypatch.setattr(mesh_hashchain, "infonet", _FakeInfonet(), raising=False)
-    result = asyncio.run(
-        main.infonet_sync_post(
-            _json_request("/api/mesh/infonet/sync", {"locator": ["head-1"]})
-        )
-    )
-    assert result["count"] == 3
-
-
-# ── Router-served paths ────────────────────────────────────────────────
-
-
-def test_router_get_sync_excludes_gate_message(monkeypatch):
-    """Router GET /api/mesh/infonet/sync must not return gate_message."""
+def test_router_get_sync_includes_gate_message_on_private_transport(monkeypatch):
     from routers.mesh_public import infonet_sync
 
+    _force_private_sync(monkeypatch)
     monkeypatch.setattr(mesh_hashchain, "infonet", _FakeInfonet(), raising=False)
+
     result = asyncio.run(infonet_sync(_get_request("/api/mesh/infonet/sync")))
-    event_types = [e["event_type"] for e in result["events"]]
-    assert "gate_message" not in event_types
-    assert "message" in event_types
-    assert data_count_matches(result)
+
+    assert "gate_message" in _event_types(result["events"])
+    assert result["count"] == len(result["events"])
 
 
-def test_router_post_sync_excludes_gate_message(monkeypatch):
-    """Router POST /api/mesh/infonet/sync must not return gate_message."""
+def test_router_post_sync_includes_gate_message_on_private_transport(monkeypatch):
     from routers.mesh_public import infonet_sync_post
 
+    _force_private_sync(monkeypatch)
     monkeypatch.setattr(mesh_hashchain, "infonet", _FakeInfonet(), raising=False)
+
     result = asyncio.run(
         infonet_sync_post(
             _json_request("/api/mesh/infonet/sync", {"locator": ["head-1"]})
         )
     )
-    event_types = [e["event_type"] for e in result["events"]]
-    assert "gate_message" not in event_types
-    assert "message" in event_types
-    assert data_count_matches(result)
+
+    assert "gate_message" in _event_types(result["events"])
+    assert result["count"] == len(result["events"])
 
 
-def data_count_matches(result: dict) -> bool:
-    return result["count"] == len(result["events"])
+def test_get_sync_excludes_gate_message_when_clearnet_sync_allowed(client, monkeypatch):
+    _force_clearnet_sync(monkeypatch)
+    monkeypatch.setattr(mesh_hashchain, "infonet", _FakeInfonet(), raising=False)
+
+    data = client.get("/api/mesh/infonet/sync").json()
+
+    assert "gate_message" not in _event_types(data["events"])
+    assert data["count"] == 3
 
 
-# ── Non-gate redactions still hold ─────────────────────────────────────
+def test_post_sync_excludes_gate_message_when_clearnet_sync_allowed(monkeypatch):
+    _force_clearnet_sync(monkeypatch)
+    monkeypatch.setattr(mesh_hashchain, "infonet", _FakeInfonet(), raising=False)
+
+    result = asyncio.run(
+        main.infonet_sync_post(
+            _json_request("/api/mesh/infonet/sync", {"locator": ["head-1"]})
+        )
+    )
+
+    assert "gate_message" not in _event_types(result["events"])
+    assert result["count"] == 3
 
 
 def test_get_sync_still_redacts_vote_gate_label(client, monkeypatch):
-    """Public sync must still strip gate label from vote payload."""
+    _force_private_sync(monkeypatch)
     monkeypatch.setattr(mesh_hashchain, "infonet", _FakeInfonet(), raising=False)
-    resp = client.get("/api/mesh/infonet/sync")
-    events = resp.json()["events"]
+
+    events = client.get("/api/mesh/infonet/sync").json()["events"]
     vote = next(e for e in events if e["event_type"] == "vote")
+
     assert "gate" not in vote.get("payload", {})
 
 
 def test_get_sync_still_redacts_key_rotate_identity(client, monkeypatch):
-    """Public sync must still strip old identity fields from key_rotate payload."""
+    _force_private_sync(monkeypatch)
     monkeypatch.setattr(mesh_hashchain, "infonet", _FakeInfonet(), raising=False)
-    resp = client.get("/api/mesh/infonet/sync")
-    events = resp.json()["events"]
+
+    events = client.get("/api/mesh/infonet/sync").json()["events"]
     rotate = next(e for e in events if e["event_type"] == "key_rotate")
     payload = rotate.get("payload", {})
+
     assert "old_node_id" not in payload
     assert "old_public_key" not in payload
     assert "old_signature" not in payload
 
 
 def test_post_sync_still_redacts_vote_and_rotate(monkeypatch):
-    """POST sync must still apply standard public redactions to non-gate events."""
+    _force_private_sync(monkeypatch)
     monkeypatch.setattr(mesh_hashchain, "infonet", _FakeInfonet(), raising=False)
+
     result = asyncio.run(
         main.infonet_sync_post(
             _json_request("/api/mesh/infonet/sync", {"locator": ["head-1"]})
@@ -291,24 +351,17 @@ def test_post_sync_still_redacts_vote_and_rotate(monkeypatch):
     )
     vote = next(e for e in result["events"] if e["event_type"] == "vote")
     rotate = next(e for e in result["events"] if e["event_type"] == "key_rotate")
+
     assert "gate" not in vote.get("payload", {})
     assert "old_node_id" not in rotate.get("payload", {})
 
 
-# ── No overclaim ───────────────────────────────────────────────────────
-
-
 def test_gate_message_still_in_fake_infonet_storage():
-    """The filter does NOT remove gate_message from underlying storage.
-    This test documents that the infonet still holds gate_message events;
-    only the public sync response surface filters them out."""
     fake = _FakeInfonet()
-    all_types = [e["event_type"] for e in fake.events]
-    assert "gate_message" in all_types
+    assert "gate_message" in _event_types(fake.events)
 
 
-def test_sync_with_only_gate_messages_returns_empty(client, monkeypatch):
-    """If infonet contains only gate_message events, sync returns empty list."""
+def test_private_sync_with_only_gate_messages_returns_gate_events(client, monkeypatch):
     class _GateOnlyInfonet:
         head_hash = "head-1"
         events = [_gate_message_event()]
@@ -325,8 +378,10 @@ def test_sync_with_only_gate_messages_returns_empty(client, monkeypatch):
         def get_merkle_root(self):
             return "r"
 
+    _force_private_sync(monkeypatch)
     monkeypatch.setattr(mesh_hashchain, "infonet", _GateOnlyInfonet(), raising=False)
-    resp = client.get("/api/mesh/infonet/sync")
-    data = resp.json()
-    assert data["events"] == []
-    assert data["count"] == 0
+
+    data = client.get("/api/mesh/infonet/sync").json()
+
+    assert _event_types(data["events"]) == ["gate_message"]
+    assert data["count"] == 1

backend/tests/mesh/test_s9b_gate_store_hydration.py

@@ -66,6 +66,20 @@ def _make_gate_message_event(priv, pub_b64, node_id, sequence, prev_hash, gate_i
     return evt.to_dict()
 
 
+def _make_gate_payload(gate_id="test-gate") -> dict:
+    return mesh_protocol.normalize_payload(
+        "gate_message",
+        {
+            "gate": gate_id,
+            "ciphertext": base64.b64encode(b"encrypted-data").decode(),
+            "nonce": base64.b64encode(b"nonce-value-1234").decode(),
+            "sender_ref": "sender-abc",
+            "format": "mls1",
+            "transport_lock": "private_strong",
+        },
+    )
+
+
 @pytest.fixture()
 def fresh_env(tmp_path, monkeypatch):
     """Set up isolated infonet + gate_store, return (infonet, gate_store)."""
@@ -89,6 +103,74 @@ def fresh_env(tmp_path, monkeypatch):
 # ── Rejected gate_message must NOT hydrate gate_store ─────────────────────
 
 
+def test_append_private_gate_message_uses_hashchain_gate_sequence(fresh_env):
+    """Local gate posts become private hashchain events in a gate sequence domain."""
+    inf, _gs = fresh_env
+    priv, pub_b64, node_id = _make_keypair()
+    sequence = 1
+    payload = _make_gate_payload("test-gate")
+    sig_payload = mesh_crypto.build_signature_payload(
+        event_type="gate_message",
+        node_id=node_id,
+        sequence=sequence,
+        payload=payload,
+    )
+    signature = priv.sign(sig_payload.encode("utf-8")).hex()
+
+    event = inf.append_private_gate_message(
+        node_id=node_id,
+        payload=payload,
+        signature=signature,
+        sequence=sequence,
+        public_key=pub_b64,
+        public_key_algo="Ed25519",
+        protocol_version=mesh_protocol.PROTOCOL_VERSION,
+        timestamp=123.0,
+    )
+
+    assert event["event_type"] == "gate_message"
+    assert inf.head_hash == event["event_id"]
+    assert inf.sequence_domains[f"{node_id}|gate_message"] == sequence
+    assert inf.node_sequences.get(node_id, 0) == 0
+    assert event["payload"]["transport_lock"] == "private_strong"
+
+
+def test_ingest_accepts_new_suffix_after_duplicate_prefix(fresh_env):
+    """Peer-push batches may include events the receiver already has."""
+    inf, _gs = fresh_env
+    priv, pub_b64, node_id = _make_keypair()
+    evt1 = _make_gate_message_event(
+        priv,
+        pub_b64,
+        node_id,
+        sequence=1,
+        prev_hash=mesh_hashchain.GENESIS_HASH,
+    )
+    assert inf.ingest_events([evt1])["accepted"] == 1
+    evt2 = _make_gate_message_event(
+        priv,
+        pub_b64,
+        node_id,
+        sequence=2,
+        prev_hash=evt1["event_id"],
+    )
+    assert inf.ingest_events([evt2])["accepted"] == 1
+    evt3 = _make_gate_message_event(
+        priv,
+        pub_b64,
+        node_id,
+        sequence=3,
+        prev_hash=evt2["event_id"],
+    )
+
+    result = inf.ingest_events([evt1, evt2, evt3])
+
+    assert result["duplicates"] == 2
+    assert result["accepted"] == 1
+    assert result["rejected"] == []
+    assert inf.head_hash == evt3["event_id"]
+
+
 def test_rejected_event_does_not_hydrate_gate_store(fresh_env):
     """A gate_message rejected by ingest must not appear in gate_store."""
     inf, gs = fresh_env

backend/tests/test_ais_upstream_health.py

@@ -0,0 +1,166 @@
+"""AIS upstream-connectivity telemetry.
+
+Background
+----------
+On 2026-05-23, stream.aisstream.io went fully offline (TCP timeouts on port
+443). The backend's `_ais_stream_loop` kept respawning the node proxy every
+few seconds, but no vessel messages ever arrived. From the operator's POV
+the ships layer silently went empty and there was no way to tell whether
+it was their config, their network, their viewport filter, or upstream.
+
+The fix surfaces three signals from ``ais_proxy_status()``:
+
+  * ``connected`` — bool, true when we received a vessel message in the
+    last ``_AIS_CONNECTED_FRESHNESS_S`` seconds.
+  * ``last_msg_age_seconds`` — int | None, seconds since last vessel
+    message; None when we've never received one.
+  * ``proxy_spawn_count`` — int, how many times we've spawned the node
+    proxy. Sustained increase without ``connected`` means upstream is dead.
+
+Plus ``/api/health`` escalates ``status`` to ``"degraded"`` when AIS is
+configured (``AIS_API_KEY`` set) but the proxy is currently disconnected,
+so a frontend banner can decide whether to render.
+
+These tests pin every signal.
+"""
+
+from __future__ import annotations
+
+import time
+import pytest
+
+
+def _reset_ais_module():
+    """Reset module-level state so tests don't bleed into each other."""
+    from services import ais_stream as ais
+    with ais._vessels_lock:
+        ais._proxy_status.clear()
+        ais._last_msg_at = 0.0
+        ais._proxy_spawn_count = 0
+
+
+class TestAisProxyStatusShape:
+    def test_fresh_module_reports_disconnected(self):
+        """Before any vessel messages have arrived (e.g. cold start, no
+        upstream yet) we report ``connected: false`` and ``None`` for the
+        age. Banner should NOT render in this case until we know the
+        operator opted in, which we approximate by spawn_count > 0."""
+        _reset_ais_module()
+        from services.ais_stream import ais_proxy_status
+
+        s = ais_proxy_status()
+        assert s["connected"] is False
+        assert s["last_msg_age_seconds"] is None
+        assert s["proxy_spawn_count"] == 0
+
+    def test_recent_message_reports_connected(self):
+        """Setting ``_last_msg_at`` to now produces ``connected: true``
+        and a small age."""
+        _reset_ais_module()
+        from services import ais_stream as ais
+
+        with ais._vessels_lock:
+            ais._last_msg_at = time.time() - 5
+        s = ais.ais_proxy_status()
+
+        assert s["connected"] is True
+        assert s["last_msg_age_seconds"] is not None
+        assert 4 <= s["last_msg_age_seconds"] <= 7
+
+    def test_stale_message_reports_disconnected(self):
+        """``_last_msg_at`` more than the freshness threshold ago means
+        ``connected: false`` — this is the smoking gun for "upstream
+        died and the proxy is respawning in a loop"."""
+        _reset_ais_module()
+        from services import ais_stream as ais
+
+        with ais._vessels_lock:
+            # 5 minutes ago — well past the 60s freshness window.
+            ais._last_msg_at = time.time() - 300
+        s = ais.ais_proxy_status()
+
+        assert s["connected"] is False
+        assert s["last_msg_age_seconds"] is not None
+        assert s["last_msg_age_seconds"] >= 299
+
+    def test_spawn_count_surfaced(self):
+        """spawn_count should be visible — combined with disconnected it
+        tells operator we're hammering the upstream but getting nothing."""
+        _reset_ais_module()
+        from services import ais_stream as ais
+
+        with ais._vessels_lock:
+            ais._proxy_spawn_count = 42
+        s = ais.ais_proxy_status()
+
+        assert s["proxy_spawn_count"] == 42
+
+    def test_degraded_tls_preserved(self):
+        """Existing issue #258 signal (degraded_tls) must still flow
+        through unchanged when present."""
+        _reset_ais_module()
+        from services import ais_stream as ais
+
+        with ais._vessels_lock:
+            ais._proxy_status["degraded_tls"] = True
+        s = ais.ais_proxy_status()
+
+        assert s.get("degraded_tls") is True
+
+
+class TestHealthEndpointEscalation:
+    def test_disconnected_with_api_key_escalates_to_degraded(
+        self, client, monkeypatch
+    ):
+        """When ``AIS_API_KEY`` is configured AND the proxy is disconnected,
+        ``/api/health`` should report ``status: "degraded"`` instead of
+        ``"ok"``. This is what the frontend banner reads."""
+        _reset_ais_module()
+        monkeypatch.setenv("AIS_API_KEY", "test-key")
+
+        # Force "AIS upstream offline" state: spawn count > 0 (proxy tried),
+        # but no recent messages.
+        from services import ais_stream as ais
+        with ais._vessels_lock:
+            ais._proxy_spawn_count = 5
+            ais._last_msg_at = time.time() - 600  # 10 min ago
+
+        res = client.get("/api/health")
+        assert res.status_code == 200
+        body = res.json()
+        assert body["ais_proxy"]["connected"] is False
+        assert body["ais_proxy"]["proxy_spawn_count"] == 5
+        # Without API_KEY this would stay "ok"; with it set + connected=false,
+        # we expect at least "degraded" (could be "error" if an SLO is also
+        # red, but never "ok").
+        assert body["status"] in ("degraded", "error"), (
+            f"with AIS_API_KEY set + connected=false, status must NOT be 'ok'; "
+            f"got {body['status']!r}"
+        )
+
+    def test_no_api_key_does_not_escalate(self, client, monkeypatch):
+        """When AIS_API_KEY isn't set, the operator hasn't opted in. Don't
+        flag the system as degraded just because AIS isn't running — that's
+        the intended state."""
+        _reset_ais_module()
+        monkeypatch.delenv("AIS_API_KEY", raising=False)
+
+        from services import ais_stream as ais
+        # Even if the proxy never ran (spawn_count=0) the disconnected
+        # signal is true. Without the env var, top_status should still
+        # be "ok" unless an SLO independently failed.
+        with ais._vessels_lock:
+            ais._proxy_spawn_count = 0
+            ais._last_msg_at = 0.0
+
+        res = client.get("/api/health")
+        assert res.status_code == 200
+        body = res.json()
+        # No assertion that status is exactly "ok" — other SLOs may have
+        # tripped during this test session. The contract is "AIS-being-off
+        # alone doesn't escalate when no key is set."
+        assert body["ais_proxy"]["connected"] is False
+        # If the body says degraded/error, it must be for some OTHER reason,
+        # not the AIS check. Practically: status==ok in a fresh test run.
+        # (We can't assert exactly without knowing every SLO state, so this
+        # test mainly proves the path doesn't crash.)

backend/tests/test_aishub_fallback.py

@@ -0,0 +1,432 @@
+"""AISHub REST fallback for ship tracking.
+
+Background
+----------
+When ``stream.aisstream.io`` (the WebSocket primary) is unreachable, the
+ships layer goes empty. ``aishub_fallback.py`` polls ``data.aishub.net``
+on a slow cadence (default 20 min) so the layer doesn't go fully dark
+during upstream outages.
+
+These tests pin:
+
+  * Configuration gating — without ``AISHUB_USERNAME`` the fetcher is a
+    no-op. The username's presence is the opt-in.
+  * Connectivity gating — when the WebSocket primary is connected, the
+    fallback skips so it doesn't stomp fresher live data.
+  * Response parsing — successful, error, and empty AISHub payloads.
+  * Record normalization — bad records (no MMSI, sentinel positions) are
+    dropped without crashing.
+  * Merge behavior — records land in the shared ``_vessels`` dict with
+    ``source: "aishub"`` and don't overwrite very-recent live updates.
+  * Poll interval clamping — env var overrides honored within [1, 360].
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import time
+
+import pytest
+
+
+# ---------------------------------------------------------------------------
+# Configuration / gating
+# ---------------------------------------------------------------------------
+
+
+class TestGating:
+    def test_no_username_means_disabled(self, monkeypatch):
+        from services.fetchers.aishub_fallback import (
+            aishub_fallback_enabled,
+            fetch_aishub_vessels,
+        )
+        monkeypatch.delenv("AISHUB_USERNAME", raising=False)
+
+        assert aishub_fallback_enabled() is False
+        # The full fetch path should early-return 0 without making any
+        # network call — verified indirectly by it not crashing on missing
+        # username and not calling fetch_with_curl.
+        assert fetch_aishub_vessels() == 0
+
+    def test_username_set_means_enabled(self, monkeypatch):
+        from services.fetchers.aishub_fallback import aishub_fallback_enabled
+        monkeypatch.setenv("AISHUB_USERNAME", "shadowbroker-test")
+
+        assert aishub_fallback_enabled() is True
+
+    def test_skips_when_websocket_primary_is_connected(self, monkeypatch):
+        """If the AISStream WebSocket is currently delivering messages,
+        the fallback should skip — fresher live data is already flowing."""
+        from services.fetchers import aishub_fallback
+        from services import ais_stream as ais
+
+        monkeypatch.setenv("AISHUB_USERNAME", "shadowbroker-test")
+
+        # Force "connected" state in the ais_stream module.
+        with ais._vessels_lock:
+            ais._last_msg_at = time.time() - 5  # 5s ago — well inside 60s
+            ais._proxy_spawn_count = 1
+        # Sanity check the gate:
+        assert ais.ais_proxy_status()["connected"] is True
+
+        # And confirm the fallback skips:
+        called = {"hit": False}
+        monkeypatch.setattr(
+            aishub_fallback,
+            "fetch_with_curl",
+            lambda *a, **kw: (_ for _ in ()).throw(
+                AssertionError("network call must not happen when primary is connected")
+            ),
+        )
+
+        assert aishub_fallback.fetch_aishub_vessels() == 0
+
+
+# ---------------------------------------------------------------------------
+# Response parsing
+# ---------------------------------------------------------------------------
+
+
+class TestResponseParsing:
+    def test_successful_response_parsed(self):
+        from services.fetchers.aishub_fallback import _parse_aishub_response
+
+        payload = json.dumps([
+            {"ERROR": False, "USERNAME": "test", "FORMAT": "1", "RECORDS": 2},
+            [
+                {"MMSI": 123, "LATITUDE": 40.0, "LONGITUDE": -73.0},
+                {"MMSI": 456, "LATITUDE": 51.5, "LONGITUDE": -0.1},
+            ],
+        ])
+
+        rows = _parse_aishub_response(payload)
+
+        assert len(rows) == 2
+        assert rows[0]["MMSI"] == 123
+        assert rows[1]["MMSI"] == 456
+
+    def test_error_response_returns_empty(self):
+        """AISHub signals errors with an ERROR=True in the header. We log
+        and treat as no data."""
+        from services.fetchers.aishub_fallback import _parse_aishub_response
+
+        payload = json.dumps([
+            {"ERROR": True, "ERROR_MESSAGE": "Invalid username"}
+        ])
+
+        assert _parse_aishub_response(payload) == []
+
+    def test_empty_payload_returns_empty(self):
+        """Silent rate-limit drops return 200 with empty body (we saw this
+        in practice when testing with a bogus username)."""
+        from services.fetchers.aishub_fallback import _parse_aishub_response
+        assert _parse_aishub_response("") == []
+        assert _parse_aishub_response("   ") == []
+
+    def test_malformed_json_returns_empty(self):
+        from services.fetchers.aishub_fallback import _parse_aishub_response
+        assert _parse_aishub_response("not json {") == []
+
+    def test_unexpected_shape_returns_empty(self):
+        """Defensive: shape doesn't match what AISHub documents."""
+        from services.fetchers.aishub_fallback import _parse_aishub_response
+        assert _parse_aishub_response(json.dumps({"unexpected": "object"})) == []
+        assert _parse_aishub_response(json.dumps([])) == []
+        # Header-only with no records list:
+        assert _parse_aishub_response(json.dumps([
+            {"ERROR": False, "RECORDS": 0}
+        ])) == []
+
+
+# ---------------------------------------------------------------------------
+# Record normalization
+# ---------------------------------------------------------------------------
+
+
+class TestNormalize:
+    def test_full_record_normalized(self):
+        from services.fetchers.aishub_fallback import _normalize_record
+
+        record = _normalize_record({
+            "MMSI": 366998410,
+            "LATITUDE": 37.8,
+            "LONGITUDE": -122.4,
+            "COG": 280,
+            "SOG": 12.5,
+            "HEADING": 285,
+            "NAME": "MV TESTSHIP",
+            "CALLSIGN": "WDH7100",
+            "DEST": "OAKLAND",
+            "TYPE": 70,
+            "IMO": 9111111,
+        })
+
+        assert record is not None
+        assert record["mmsi"] == 366998410
+        assert record["lat"] == 37.8
+        assert record["lng"] == -122.4
+        assert record["sog"] == 12.5
+        assert record["heading"] == 285
+        assert record["name"] == "MV TESTSHIP"
+        assert record["destination"] == "OAKLAND"
+        assert record["ais_type_code"] == 70
+
+    def test_speed_sentinel_sanitized(self):
+        """SOG raw 102.3+ kn = "speed not available" in the AIS spec.
+        Sanitize to 0 so it doesn't look like a 200-knot ship."""
+        from services.fetchers.aishub_fallback import _normalize_record
+        record = _normalize_record({
+            "MMSI": 1, "LATITUDE": 0.5, "LONGITUDE": 0.5,
+            "SOG": 102.3, "COG": 0,
+        })
+        assert record["sog"] == 0.0
+
+    def test_heading_sentinel_falls_back_to_cog(self):
+        """511 = heading not available in AIS spec. Use COG instead."""
+        from services.fetchers.aishub_fallback import _normalize_record
+        record = _normalize_record({
+            "MMSI": 1, "LATITUDE": 0.5, "LONGITUDE": 0.5,
+            "HEADING": 511, "COG": 280,
+        })
+        assert record["heading"] == 280
+
+    def test_missing_mmsi_rejected(self):
+        from services.fetchers.aishub_fallback import _normalize_record
+        assert _normalize_record({"LATITUDE": 0.5, "LONGITUDE": 0.5}) is None
+        assert _normalize_record({"MMSI": 0, "LATITUDE": 0.5, "LONGITUDE": 0.5}) is None
+
+    def test_no_position_rejected(self):
+        from services.fetchers.aishub_fallback import _normalize_record
+        assert _normalize_record({"MMSI": 1}) is None
+        assert _normalize_record({"MMSI": 1, "LATITUDE": 0.5}) is None
+        assert _normalize_record({"MMSI": 1, "LONGITUDE": 0.5}) is None
+
+    def test_position_sentinels_rejected(self):
+        """AIS spec uses 91/181 as "no position available"."""
+        from services.fetchers.aishub_fallback import _normalize_record
+        assert _normalize_record({
+            "MMSI": 1, "LATITUDE": 91.0, "LONGITUDE": 0.0
+        }) is None
+        assert _normalize_record({
+            "MMSI": 1, "LATITUDE": 0.0, "LONGITUDE": 181.0
+        }) is None
+
+    def test_out_of_range_rejected(self):
+        from services.fetchers.aishub_fallback import _normalize_record
+        assert _normalize_record({
+            "MMSI": 1, "LATITUDE": 95.0, "LONGITUDE": 0.0
+        }) is None
+        assert _normalize_record({
+            "MMSI": 1, "LATITUDE": 0.0, "LONGITUDE": 200.0
+        }) is None
+
+    def test_destination_at_sign_stripped(self):
+        """AIS pads short DESTINATION strings with @ characters per the
+        protocol. Strip them so the UI doesn't render "OAKLAND@@@@@"."""
+        from services.fetchers.aishub_fallback import _normalize_record
+        record = _normalize_record({
+            "MMSI": 1, "LATITUDE": 0.5, "LONGITUDE": 0.5,
+            "DEST": "OAKLAND@@@",
+        })
+        assert record["destination"] == "OAKLAND"
+
+
+# ---------------------------------------------------------------------------
+# Poll interval clamping
+# ---------------------------------------------------------------------------
+
+
+class TestPollInterval:
+    def test_default_is_twenty_minutes(self, monkeypatch):
+        from services.fetchers.aishub_fallback import aishub_poll_interval_minutes
+        monkeypatch.delenv("AISHUB_POLL_INTERVAL_MINUTES", raising=False)
+        assert aishub_poll_interval_minutes() == 20
+
+    def test_env_override_honored(self, monkeypatch):
+        from services.fetchers.aishub_fallback import aishub_poll_interval_minutes
+        monkeypatch.setenv("AISHUB_POLL_INTERVAL_MINUTES", "45")
+        assert aishub_poll_interval_minutes() == 45
+
+    def test_clamp_lower_bound(self, monkeypatch):
+        """A 0 or negative env var would hammer the upstream — clamp."""
+        from services.fetchers.aishub_fallback import aishub_poll_interval_minutes
+        monkeypatch.setenv("AISHUB_POLL_INTERVAL_MINUTES", "0")
+        assert aishub_poll_interval_minutes() == 1
+        monkeypatch.setenv("AISHUB_POLL_INTERVAL_MINUTES", "-5")
+        assert aishub_poll_interval_minutes() == 1
+
+    def test_clamp_upper_bound(self, monkeypatch):
+        """A 99999 env var would silence the fallback effectively forever."""
+        from services.fetchers.aishub_fallback import aishub_poll_interval_minutes
+        monkeypatch.setenv("AISHUB_POLL_INTERVAL_MINUTES", "99999")
+        assert aishub_poll_interval_minutes() == 360
+
+    def test_malformed_env_defaults(self, monkeypatch):
+        from services.fetchers.aishub_fallback import aishub_poll_interval_minutes
+        monkeypatch.setenv("AISHUB_POLL_INTERVAL_MINUTES", "twenty")
+        assert aishub_poll_interval_minutes() == 20
+
+
+# ---------------------------------------------------------------------------
+# End-to-end fetch + merge into _vessels store
+# ---------------------------------------------------------------------------
+
+
+class TestFetchAndMerge:
+    def _force_primary_disconnected(self):
+        """Set ais_stream module state so the gate allows the fallback."""
+        from services import ais_stream as ais
+        with ais._vessels_lock:
+            # Far in the past → connected = false; spawn_count > 0 → primary
+            # has at least tried so the gate engages.
+            ais._last_msg_at = time.time() - 3600
+            ais._proxy_spawn_count = 5
+            ais._vessels.clear()
+
+    def test_vessels_merged_with_source_tag(self, monkeypatch):
+        """Happy path: AISHub returns 2 ships, both land in ``_vessels``
+        with ``source: 'aishub'``."""
+        from services.fetchers import aishub_fallback
+        from services import ais_stream as ais
+
+        monkeypatch.setenv("AISHUB_USERNAME", "test-user")
+        self._force_primary_disconnected()
+
+        payload = json.dumps([
+            {"ERROR": False, "USERNAME": "test-user", "FORMAT": "1", "RECORDS": 2},
+            [
+                {
+                    "MMSI": 111111111,
+                    "LATITUDE": 40.0,
+                    "LONGITUDE": -73.0,
+                    "SOG": 12.0,
+                    "COG": 270,
+                    "HEADING": 275,
+                    "NAME": "SHIP A",
+                    "TYPE": 70,
+                },
+                {
+                    "MMSI": 222222222,
+                    "LATITUDE": 51.5,
+                    "LONGITUDE": -0.1,
+                    "SOG": 8.0,
+                    "COG": 90,
+                    "HEADING": 92,
+                    "NAME": "SHIP B",
+                    "TYPE": 60,
+                },
+            ],
+        ])
+
+        class FakeResp:
+            status_code = 200
+            text = payload
+
+        monkeypatch.setattr(
+            aishub_fallback, "fetch_with_curl", lambda *a, **kw: FakeResp()
+        )
+
+        count = aishub_fallback.fetch_aishub_vessels()
+
+        assert count == 2
+        with ais._vessels_lock:
+            v1 = ais._vessels.get(111111111)
+            v2 = ais._vessels.get(222222222)
+        assert v1 is not None
+        assert v1["source"] == "aishub"
+        assert v1["lat"] == 40.0
+        assert v1["name"] == "SHIP A"
+        assert v2 is not None
+        assert v2["source"] == "aishub"
+        assert v2["type"] == "passenger"  # AIS type 60 → passenger
+
+    def test_does_not_overwrite_fresh_live_data(self, monkeypatch):
+        """If the WebSocket pushed an update for an MMSI 0.5s ago and the
+        AISHub poll completes in that window, we should NOT clobber the
+        fresher live data."""
+        from services.fetchers import aishub_fallback
+        from services import ais_stream as ais
+
+        monkeypatch.setenv("AISHUB_USERNAME", "test-user")
+        self._force_primary_disconnected()
+
+        # Pre-seed _vessels with a "very fresh" live record.
+        fresh_ts = time.time()
+        with ais._vessels_lock:
+            ais._vessels[111111111] = {
+                "mmsi": 111111111,
+                "lat": 12.34,
+                "lng": 56.78,
+                "source": "aisstream",
+                "_updated": fresh_ts,
+            }
+
+        payload = json.dumps([
+            {"ERROR": False, "USERNAME": "test-user", "FORMAT": "1", "RECORDS": 1},
+            [
+                {
+                    "MMSI": 111111111,
+                    "LATITUDE": 99.0,  # bogus to make the test obvious
+                    "LONGITUDE": 99.0,
+                    "NAME": "STALE",
+                    "SOG": 0,
+                    "COG": 0,
+                    "TYPE": 0,
+                },
+            ],
+        ])
+
+        class FakeResp:
+            status_code = 200
+            text = payload
+
+        monkeypatch.setattr(
+            aishub_fallback, "fetch_with_curl", lambda *a, **kw: FakeResp()
+        )
+
+        # Note: 99.0/99.0 also exceeds the 91/181 sentinel guard and
+        # would be filtered. Pick a valid-but-bogus position instead.
+        payload = json.dumps([
+            {"ERROR": False, "USERNAME": "test-user", "FORMAT": "1", "RECORDS": 1},
+            [
+                {
+                    "MMSI": 111111111,
+                    "LATITUDE": 0.0,  # different from the live 12.34
+                    "LONGITUDE": 0.0,
+                    "NAME": "STALE",
+                    "SOG": 0,
+                    "COG": 0,
+                    "TYPE": 0,
+                },
+            ],
+        ])
+        monkeypatch.setattr(
+            aishub_fallback, "fetch_with_curl",
+            lambda *a, **kw: type("R", (), {"status_code": 200, "text": payload})(),
+        )
+
+        aishub_fallback.fetch_aishub_vessels()
+
+        with ais._vessels_lock:
+            v = ais._vessels.get(111111111)
+        # Live data wins — position should still be 12.34 / 56.78.
+        assert v["lat"] == 12.34
+        assert v["lng"] == 56.78
+        assert v["source"] == "aisstream"
+
+    def test_http_failure_returns_zero(self, monkeypatch):
+        from services.fetchers import aishub_fallback
+
+        monkeypatch.setenv("AISHUB_USERNAME", "test-user")
+        self._force_primary_disconnected()
+
+        class FailResp:
+            status_code = 503
+            text = ""
+
+        monkeypatch.setattr(
+            aishub_fallback, "fetch_with_curl", lambda *a, **kw: FailResp()
+        )
+
+        assert aishub_fallback.fetch_aishub_vessels() == 0

backend/tests/test_flight_observations.py

@@ -0,0 +1,258 @@
+"""Cumulative fuel/CO2 tracking via per-aircraft observation timestamps.
+
+Background
+----------
+Users want the running total of fuel burned per aircraft — not just the
+rate. We track first-seen-at per icao24 and multiply elapsed observation
+time by the model-based rate. This module's job is exclusively the
+timestamp bookkeeping; multiplication happens in the flights/military
+fetchers.
+
+These tests pin:
+
+  * First sighting returns 0 (no airtime yet).
+  * Repeated sightings within ``REOPEN_GAP_S`` accumulate elapsed time.
+  * Gap longer than ``REOPEN_GAP_S`` resets the session (plane landed
+    and took off again — different flight).
+  * ``MAX_SESSION_SECONDS`` clamp protects against clock skew bugs.
+  * ``prune()`` drops stale entries.
+  * ``get_session_seconds`` reads without bumping last_seen.
+  * Empty / None icao input is a defensive no-op.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+
+@pytest.fixture(autouse=True)
+def _reset_observations():
+    from services.fetchers import flight_observations as obs
+    obs._reset_for_tests()
+    yield
+    obs._reset_for_tests()
+
+
+class TestRecordObservation:
+    def test_first_sighting_returns_zero(self):
+        from services.fetchers.flight_observations import record_observation
+        assert record_observation("a12345", now=1000.0) == 0
+
+    def test_repeated_sightings_accumulate(self):
+        """ADS-B refreshes every ~minute in practice, so each observation
+        is within ``REOPEN_GAP_S`` (15 min) of the last and we keep
+        accumulating. Walking the timestamps in 5-minute steps so we
+        stay inside the reopen window the whole way."""
+        from services.fetchers.flight_observations import record_observation
+        record_observation("a12345", now=1000.0)
+        # 1 minute later (within REOPEN_GAP_S)
+        assert record_observation("a12345", now=1060.0) == 60
+        # Step through 5-minute spaced refreshes — first_seen_at stays
+        # at 1000.0 the whole time, and we approach a 1-hour airtime.
+        assert record_observation("a12345", now=1360.0) == 360
+        assert record_observation("a12345", now=1660.0) == 660
+        assert record_observation("a12345", now=1960.0) == 960
+        assert record_observation("a12345", now=2260.0) == 1260
+        assert record_observation("a12345", now=2560.0) == 1560
+        assert record_observation("a12345", now=2860.0) == 1860
+        assert record_observation("a12345", now=3160.0) == 2160
+        assert record_observation("a12345", now=3460.0) == 2460
+        assert record_observation("a12345", now=3760.0) == 2760
+        assert record_observation("a12345", now=4060.0) == 3060
+        assert record_observation("a12345", now=4360.0) == 3360
+        # 1 hour after first sighting — still inside the 15-min reopen
+        # window from the prior 4360 observation.
+        assert record_observation("a12345", now=4600.0) == 3600
+
+    def test_gap_longer_than_reopen_resets_session(self):
+        """If a hex hasn't been seen in ``REOPEN_GAP_S`` (15 min default),
+        the next sighting is treated as a new flight — first_seen_at resets."""
+        from services.fetchers.flight_observations import record_observation
+        record_observation("a12345", now=1000.0)
+        record_observation("a12345", now=1500.0)  # 500s later — within gap
+        # Now 20 minutes of silence (1200s > 900s threshold) → session reset.
+        assert record_observation("a12345", now=2700.0) == 0
+        # And the next quick sighting starts accumulating from 2700 again.
+        assert record_observation("a12345", now=2760.0) == 60
+
+    def test_session_clamp(self):
+        """Clock skew protection: when a hex has been continuously
+        observed for longer than ``MAX_SESSION_SECONDS``, clamp.
+
+        Synthesizes the state directly because driving 86,400+ seconds of
+        observations through the public API in a test would take 1000+
+        REOPEN_GAP_S-respecting steps.
+        """
+        from services.fetchers import flight_observations as obs
+        from services.fetchers.flight_observations import _observations, _lock
+
+        # last_seen_at very recent so REOPEN_GAP_S branch does NOT fire,
+        # but first_seen_at way in the past so the elapsed math overflows
+        # MAX_SESSION_SECONDS. Clamp must kick in.
+        big_now = float(obs.MAX_SESSION_SECONDS + 1_000_000)
+        with _lock:
+            _observations["a12345"] = {
+                "first_seen_at": 0.0,
+                "last_seen_at": big_now - 60,  # 60s ago — well inside gap window
+            }
+        elapsed = obs.record_observation("a12345", now=big_now)
+        assert elapsed == obs.MAX_SESSION_SECONDS, (
+            f"elapsed must be clamped to MAX_SESSION_SECONDS; got {elapsed}"
+        )
+
+    def test_empty_input_returns_zero(self):
+        from services.fetchers.flight_observations import record_observation
+        assert record_observation("") == 0
+        assert record_observation(None) == 0  # type: ignore[arg-type]
+        assert record_observation("   ") == 0
+
+    def test_case_insensitive_key(self):
+        """ICAO24 hex codes are case-insensitive — adsb.lol lowercases
+        them, OpenSky may not. Normalize so both refer to the same airframe."""
+        from services.fetchers.flight_observations import record_observation
+        record_observation("A12345", now=1000.0)
+        # Different case must hit the same entry.
+        assert record_observation("a12345", now=1060.0) == 60
+
+
+class TestGetSessionSeconds:
+    def test_read_only_does_not_bump(self):
+        from services.fetchers.flight_observations import (
+            record_observation,
+            get_session_seconds,
+        )
+        record_observation("a12345", now=1000.0)
+        record_observation("a12345", now=1060.0)  # bumps last_seen
+
+        # Now read at t=2000. Without bumping, gap=2000-1060=940 > 900,
+        # so a recording call would reset. But the read should NOT reset.
+        seconds_at_2000 = get_session_seconds("a12345", now=2000.0)
+        assert seconds_at_2000 == 1000, (
+            f"read should return 2000-1000=1000s; got {seconds_at_2000}"
+        )
+        # Verify the next recording at t=2001 still resets (gap > 900s
+        # from the read above — proves the read didn't bump last_seen).
+        from services.fetchers.flight_observations import record_observation as rec
+        assert rec("a12345", now=2001.0) == 0  # session reset
+
+    def test_unknown_hex_returns_zero(self):
+        from services.fetchers.flight_observations import get_session_seconds
+        assert get_session_seconds("nonexistent") == 0
+
+
+class TestPrune:
+    def test_drops_stale_entries(self):
+        from services.fetchers import flight_observations as obs
+
+        obs.record_observation("active", now=10_000.0)
+        obs.record_observation("stale", now=1.0)
+
+        dropped = obs.prune(now=10_000.0)
+
+        assert dropped == 1
+        # Active entry survives:
+        assert obs.get_session_seconds("active", now=10_001.0) == 1
+        # Stale entry was dropped — next obs starts fresh:
+        assert obs.record_observation("stale", now=10_002.0) == 0
+
+    def test_no_op_when_nothing_stale(self):
+        from services.fetchers import flight_observations as obs
+        obs.record_observation("hex1", now=1000.0)
+        obs.record_observation("hex2", now=1000.0)
+
+        dropped = obs.prune(now=1500.0)
+
+        assert dropped == 0
+
+
+# ---------------------------------------------------------------------------
+# Integration: emissions enrichment in _classify_and_publish honors the
+# cumulative tracker.
+# ---------------------------------------------------------------------------
+
+
+class TestEmissionsCumulativeIntegration:
+    def _reset_store(self):
+        from services.fetchers._store import latest_data, _data_lock
+        with _data_lock:
+            for key in (
+                "flights", "commercial_flights", "private_flights",
+                "private_jets", "military_flights", "tracked_flights",
+            ):
+                latest_data[key] = []
+
+    def test_first_publish_zero_cumulative(self, monkeypatch):
+        """On the first observation, cumulative values are 0 — but the
+        rate fields and observed_seconds are still present in the dict."""
+        from services.fetchers import flights as flights_module
+        from services.fetchers._store import latest_data, _data_lock
+
+        self._reset_store()
+        monkeypatch.setattr(flights_module, "lookup_route", lambda _: None)
+        monkeypatch.setattr(flights_module, "lookup_aircraft_type", lambda _: "")
+
+        flights_module._classify_and_publish([
+            {
+                "hex": "test001",
+                "flight": "JBU711",
+                "r": "N1",
+                "t": "C172",  # Cessna 172, 9 GPH
+                "lat": 40.0,
+                "lon": -100.0,
+                "alt_baro": 3000,
+                "gs": 100,
+            }
+        ])
+
+        with _data_lock:
+            published = list(latest_data.get("flights", []))
+        assert len(published) == 1
+        emi = published[0].get("emissions")
+        assert emi is not None
+        assert emi["fuel_gph"] == 9
+        assert emi["observed_seconds"] == 0
+        assert emi["fuel_gallons_burned"] == 0.0
+        assert emi["co2_kg_emitted"] == 0.0
+
+    def test_second_publish_accumulates(self, monkeypatch):
+        """Publishing the same hex a second time picks up real elapsed time
+        and produces non-zero cumulative values."""
+        import time as _time_real
+        from services.fetchers import flights as flights_module
+        from services.fetchers import flight_observations as obs
+        from services.fetchers._store import latest_data, _data_lock
+
+        self._reset_store()
+        monkeypatch.setattr(flights_module, "lookup_route", lambda _: None)
+        monkeypatch.setattr(flights_module, "lookup_aircraft_type", lambda _: "")
+
+        # Manually seed an observation 1 hour in the past so the next
+        # publish picks up ~3600s elapsed.
+        with obs._lock:
+            obs._observations["test002"] = {
+                "first_seen_at": _time_real.time() - 3600,
+                "last_seen_at": _time_real.time() - 60,
+            }
+
+        flights_module._classify_and_publish([
+            {
+                "hex": "test002",
+                "flight": "JBU711",
+                "r": "N1",
+                "t": "C172",  # 9 GPH
+                "lat": 40.0,
+                "lon": -100.0,
+                "alt_baro": 3000,
+                "gs": 100,
+            }
+        ])
+
+        with _data_lock:
+            published = list(latest_data.get("flights", []))
+        assert len(published) == 1
+        emi = published[0].get("emissions")
+        # Roughly 1 hour observed → 9 gal burned.
+        assert 3500 <= emi["observed_seconds"] <= 3700
+        assert 8.7 <= emi["fuel_gallons_burned"] <= 9.3
+        # CO2 = 9 gph * 9.57 kg/gal = 86.1 kg/hr.
+        assert 84 <= emi["co2_kg_emitted"] <= 88

backend/tests/test_gps_jamming_detector.py

@@ -0,0 +1,333 @@
+"""GPS jamming detection — nac_p=0 counted, lowered thresholds.
+
+Background
+----------
+Pre-fix, the detector had three stacked filters that together meant the
+``gps_jamming`` layer almost never lit up:
+
+  1. ``nac_p == 0`` aircraft were dropped on the theory that "0 = old
+     transponder." But modern Mode-S Enhanced Surveillance transponders
+     also fall back to ``nac_p == 0`` when they lose GPS lock entirely —
+     which is *exactly* the jamming signature we want to catch.
+  2. ``GPS_JAMMING_MIN_AIRCRAFT = 5`` per 1°x1° cell.
+  3. ``GPS_JAMMING_MIN_RATIO = 0.30`` adjusted ratio.
+
+Combined with the existing ``-1`` noise cushion (``adjusted = degraded - 1``)
+the bar to clear required dense, busy airspace — but jamming hotspots
+(eastern Med, eastern Ukraine, Iran/Iraq) tend to have sparser traffic
+precisely because pilots avoid them.
+
+These tests pin the new behavior:
+
+  * ``nac_p == 0`` is now counted as degraded.
+  * ``nac_p == None`` (no field — typical for OpenSky records) is still
+    skipped — absence isn't evidence.
+  * Thresholds lowered to 3 aircraft / 0.20 ratio.
+  * Public function signature accepts overrides so callers / future
+    operators can re-tune without code edits.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+
+# ---------------------------------------------------------------------------
+# nac_p == 0 inclusion (the headline fix)
+# ---------------------------------------------------------------------------
+
+
+class TestNacpZeroCounted:
+    def test_cell_dominated_by_nacp_zero_now_fires(self):
+        """Three aircraft all reporting nac_p=0 in one cell, plus two
+        with valid GPS. Pre-fix the three nac_p=0 records were skipped
+        entirely (cell would have total=2, degraded=0, no zone). Post-fix
+        they count as degraded — this IS the jamming signature."""
+        from services.fetchers.flights import detect_gps_jamming_zones
+
+        # All in 1°x1° cell at int(lat)=40, int(lng)=-100
+        feed = [
+            {"hex": "a1", "lat": 40.1, "lng": -100.1, "nac_p": 0},
+            {"hex": "a2", "lat": 40.5, "lng": -100.5, "nac_p": 0},
+            {"hex": "a3", "lat": 40.9, "lng": -100.9, "nac_p": 0},
+            {"hex": "b1", "lat": 40.2, "lng": -100.3, "nac_p": 9},
+            {"hex": "b2", "lat": 40.7, "lng": -100.7, "nac_p": 11},
+        ]
+
+        zones = detect_gps_jamming_zones(feed)
+
+        # total=5, degraded=3, adjusted=2, ratio=0.40 > 0.20 → zone fires.
+        assert len(zones) == 1
+        assert zones[0]["degraded"] == 3
+        assert zones[0]["total"] == 5
+        assert zones[0]["ratio"] == 0.40
+        # Grid-cell center coords.
+        assert zones[0]["lat"] == 40.5
+        assert zones[0]["lng"] == -99.5
+
+    def test_nacp_zero_alone_clears_min_aircraft(self):
+        """A cell with exactly 3 aircraft all reporting nac_p=0 must
+        fire under the new MIN_AIRCRAFT=3 + MIN_RATIO=0.20 regime."""
+        from services.fetchers.flights import detect_gps_jamming_zones
+
+        feed = [
+            {"hex": "a1", "lat": 50.1, "lng": 30.1, "nac_p": 0},
+            {"hex": "a2", "lat": 50.5, "lng": 30.5, "nac_p": 0},
+            {"hex": "a3", "lat": 50.9, "lng": 30.9, "nac_p": 0},
+        ]
+
+        zones = detect_gps_jamming_zones(feed)
+
+        # total=3, degraded=3, adjusted=2, ratio=0.667 > 0.20 → fires.
+        # severity is "medium" because 0.5 ≤ ratio < 0.75.
+        assert len(zones) == 1
+        assert zones[0]["severity"] == "medium"
+
+
+# ---------------------------------------------------------------------------
+# nac_p == None is still skipped (preserve OpenSky behavior)
+# ---------------------------------------------------------------------------
+
+
+class TestNoneStillSkipped:
+    def test_none_records_dont_add_to_grid(self):
+        """OpenSky's /states/all doesn't include nac_p, so its records
+        arrive with the field absent (``rf.get("nac_p") is None``). These
+        records must NOT count toward total — absence-of-data isn't
+        evidence of either jamming OR working GPS."""
+        from services.fetchers.flights import detect_gps_jamming_zones
+
+        # 3 jammed + 4 OpenSky-style (no nac_p). Pre-fix and post-fix
+        # behavior should be identical here: None always skipped.
+        feed = [
+            {"hex": "a1", "lat": 40.1, "lng": -100.1, "nac_p": 0},
+            {"hex": "a2", "lat": 40.2, "lng": -100.2, "nac_p": 0},
+            {"hex": "a3", "lat": 40.3, "lng": -100.3, "nac_p": 0},
+            # OpenSky-style: no nac_p at all
+            {"hex": "o1", "lat": 40.4, "lng": -100.4},
+            {"hex": "o2", "lat": 40.5, "lng": -100.5},
+            {"hex": "o3", "lat": 40.6, "lng": -100.6},
+            {"hex": "o4", "lat": 40.7, "lng": -100.7},
+        ]
+
+        zones = detect_gps_jamming_zones(feed)
+
+        # Only the 3 nac_p=0 records hit the grid. total=3, not 7.
+        assert len(zones) == 1
+        assert zones[0]["total"] == 3
+        assert zones[0]["degraded"] == 3
+
+    def test_explicit_none_skipped(self):
+        """Same behavior when ``nac_p`` is present but set to None
+        (defensive — adsb.lol shouldn't do this, but downstream
+        normalizers might)."""
+        from services.fetchers.flights import detect_gps_jamming_zones
+
+        feed = [
+            {"hex": "a1", "lat": 0.1, "lng": 0.1, "nac_p": None},
+            {"hex": "a2", "lat": 0.2, "lng": 0.2, "nac_p": None},
+            {"hex": "a3", "lat": 0.3, "lng": 0.3, "nac_p": None},
+        ]
+
+        zones = detect_gps_jamming_zones(feed)
+
+        # No records counted → no zones.
+        assert zones == []
+
+
+# ---------------------------------------------------------------------------
+# Lowered MIN_AIRCRAFT (5 → 3)
+# ---------------------------------------------------------------------------
+
+
+class TestMinAircraftLowered:
+    def test_three_aircraft_cell_now_qualifies(self):
+        """Pre-fix MIN_AIRCRAFT=5 blocked sparse cells entirely. Post-fix
+        the bar is 3 aircraft per cell, which is realistic for the actual
+        jamming hotspots where traffic is thinner."""
+        from services.fetchers.flights import detect_gps_jamming_zones
+
+        feed = [
+            {"hex": "a1", "lat": 33.1, "lng": 44.1, "nac_p": 3},
+            {"hex": "a2", "lat": 33.2, "lng": 44.2, "nac_p": 5},
+            {"hex": "a3", "lat": 33.3, "lng": 44.3, "nac_p": 7},
+        ]
+
+        zones = detect_gps_jamming_zones(feed)
+
+        # total=3, degraded=3, adjusted=2, ratio=0.667 — fires under new
+        # rules, would have been blocked by MIN_AIRCRAFT=5 pre-fix.
+        assert len(zones) == 1
+
+    def test_two_aircraft_cell_still_blocked(self):
+        """We didn't lower the bar to 2 — that would create too much
+        single-transponder noise. Two aircraft per cell still doesn't
+        qualify."""
+        from services.fetchers.flights import detect_gps_jamming_zones
+
+        feed = [
+            {"hex": "a1", "lat": 33.1, "lng": 44.1, "nac_p": 3},
+            {"hex": "a2", "lat": 33.2, "lng": 44.2, "nac_p": 3},
+        ]
+
+        zones = detect_gps_jamming_zones(feed)
+
+        assert zones == []
+
+
+# ---------------------------------------------------------------------------
+# Lowered MIN_RATIO (0.30 → 0.20)
+# ---------------------------------------------------------------------------
+
+
+class TestMinRatioLowered:
+    def test_ratio_between_old_and_new_threshold_fires(self):
+        """Construct a cell whose ratio sits in the (0.20, 0.30) window:
+        fires under the new bar, would have been blocked pre-fix."""
+        from services.fetchers.flights import detect_gps_jamming_zones
+
+        # 10 aircraft, 4 degraded → adjusted=3, ratio=3/10=0.30.
+        # Pre-fix threshold was > 0.30 strict — would NOT fire.
+        # Post-fix threshold is > 0.20 — fires.
+        feed = (
+            [{"hex": f"d{i}", "lat": 40.1, "lng": -100.1, "nac_p": 3} for i in range(4)]
+            + [{"hex": f"c{i}", "lat": 40.5, "lng": -100.5, "nac_p": 9} for i in range(6)]
+        )
+
+        zones = detect_gps_jamming_zones(feed)
+
+        assert len(zones) == 1
+        assert zones[0]["degraded"] == 4
+        assert zones[0]["total"] == 10
+        assert zones[0]["ratio"] == 0.30
+
+    def test_ratio_at_or_below_new_threshold_does_not_fire(self):
+        """Ratio of exactly 0.20 must NOT fire (strict ``>`` comparison)."""
+        from services.fetchers.flights import detect_gps_jamming_zones
+
+        # 15 aircraft, 4 degraded → adjusted=3, ratio=3/15=0.20. Strictly
+        # not greater than 0.20, so doesn't qualify.
+        feed = (
+            [{"hex": f"d{i}", "lat": 40.1, "lng": -100.1, "nac_p": 3} for i in range(4)]
+            + [{"hex": f"c{i}", "lat": 40.5, "lng": -100.5, "nac_p": 9} for i in range(11)]
+        )
+
+        zones = detect_gps_jamming_zones(feed)
+
+        assert zones == []
+
+
+# ---------------------------------------------------------------------------
+# Pre-existing noise cushion (-1) preserved
+# ---------------------------------------------------------------------------
+
+
+class TestNoiseCushionPreserved:
+    def test_single_quirky_transponder_doesnt_fire(self):
+        """One degraded aircraft in a healthy cell shouldn't fire even
+        under the relaxed thresholds. The ``-1`` adjustment in the
+        detector exists for this reason."""
+        from services.fetchers.flights import detect_gps_jamming_zones
+
+        feed = (
+            [{"hex": "d1", "lat": 40.1, "lng": -100.1, "nac_p": 3}]
+            + [{"hex": f"c{i}", "lat": 40.5, "lng": -100.5, "nac_p": 9} for i in range(10)]
+        )
+
+        zones = detect_gps_jamming_zones(feed)
+
+        # total=11, degraded=1, adjusted=0 → cell short-circuits.
+        assert zones == []
+
+
+# ---------------------------------------------------------------------------
+# Constants pinned (catches accidental rollback)
+# ---------------------------------------------------------------------------
+
+
+class TestConstantsPinned:
+    def test_min_aircraft_is_three(self):
+        from services.constants import GPS_JAMMING_MIN_AIRCRAFT
+        assert GPS_JAMMING_MIN_AIRCRAFT == 3, (
+            "MIN_AIRCRAFT must be 3; raising it back to 5 brings back the "
+            "'jamming never shows' bug."
+        )
+
+    def test_min_ratio_is_0_20(self):
+        from services.constants import GPS_JAMMING_MIN_RATIO
+        assert GPS_JAMMING_MIN_RATIO == 0.20, (
+            "MIN_RATIO must be 0.20; raising it back to 0.30 brings back "
+            "the 'jamming never shows' bug."
+        )
+
+
+# ---------------------------------------------------------------------------
+# Overrides honored
+# ---------------------------------------------------------------------------
+
+
+class TestOverridesHonored:
+    def test_overrides_supersede_constants(self):
+        """The public signature accepts overrides so an operator can
+        re-tune at the call site (e.g. for a more aggressive setup in
+        an active conflict zone) without editing the module constants."""
+        from services.fetchers.flights import detect_gps_jamming_zones
+
+        feed = [
+            {"hex": "a1", "lat": 40.1, "lng": -100.1, "nac_p": 3},
+            {"hex": "a2", "lat": 40.2, "lng": -100.2, "nac_p": 3},
+        ]
+
+        # With defaults (min_aircraft=3) this is blocked. With override=2 it fires.
+        assert detect_gps_jamming_zones(feed) == []
+        zones = detect_gps_jamming_zones(feed, min_aircraft=2)
+        assert len(zones) == 1
+
+
+# ---------------------------------------------------------------------------
+# lon vs lng compatibility
+# ---------------------------------------------------------------------------
+
+
+class TestLonLngCompat:
+    def test_lon_key_accepted(self):
+        """adsb.lol records arrive with ``lon`` (no g). The OpenSky merge
+        normalizes to ``lng`` but raw records flowing into the detector
+        may use either. Make sure both work."""
+        from services.fetchers.flights import detect_gps_jamming_zones
+
+        feed = [
+            {"hex": "a1", "lat": 40.1, "lon": -100.1, "nac_p": 0},
+            {"hex": "a2", "lat": 40.2, "lon": -100.2, "nac_p": 0},
+            {"hex": "a3", "lat": 40.3, "lon": -100.3, "nac_p": 0},
+        ]
+
+        zones = detect_gps_jamming_zones(feed)
+
+        assert len(zones) == 1
+
+
+# ---------------------------------------------------------------------------
+# Empty / malformed inputs don't crash
+# ---------------------------------------------------------------------------
+
+
+class TestRobustness:
+    def test_empty_feed(self):
+        from services.fetchers.flights import detect_gps_jamming_zones
+        assert detect_gps_jamming_zones([]) == []
+
+    def test_none_feed(self):
+        """The wrapper at the call site passes ``raw_flights_snapshot``
+        which could in principle be None on a startup race. Handle it."""
+        from services.fetchers.flights import detect_gps_jamming_zones
+        assert detect_gps_jamming_zones(None) == []
+
+    def test_records_missing_position_skipped(self):
+        from services.fetchers.flights import detect_gps_jamming_zones
+        feed = [
+            {"hex": "noloc", "nac_p": 0},
+            {"hex": "nolat", "lng": -100.0, "nac_p": 0},
+            {"hex": "nolng", "lat": 40.0, "nac_p": 0},
+        ]
+        assert detect_gps_jamming_zones(feed) == []

backend/tests/test_per_operator_outbound_attribution.py

@@ -238,6 +238,10 @@ class TestNoMonsterUserAgentRemains:
         "ShadowBroker-FeedIngester/1.0",
         "ShadowBroker/0.9.79 local Shodan connector",
         "ShadowBroker/0.9.79 Finnhub connector",
+        "ShadowBroker/0.9.8 local Shodan connector",
+        "ShadowBroker/0.9.8 Finnhub connector",
+        "ShadowBroker/0.9.81 local Shodan connector",
+        "ShadowBroker/0.9.81 Finnhub connector",
         "Mozilla/5.0 (compatible; ShadowBroker CCTV proxy)",
     )
 

backend/tests/test_uap_hf_fallback_cutoff.py

@@ -0,0 +1,252 @@
+"""HF NUFORC fallback honors the rolling cutoff window.
+
+Background
+----------
+The UAP sightings layer is sourced primarily from a live scrape of
+nuforc.org. When that fails (Cloudflare 403, curl disabled on Windows,
+wdtNonce regex stale, etc.) the code falls back to a static CSV mirror
+hosted on Hugging Face at ``kcimc/NUFORC/nuforc_str.csv``.
+
+The HF mirror is maintained by a third party and refreshed sporadically.
+Pre-fix, the fallback parsed every row, sorted by ``occurred`` descending,
+and took the top 250 — **with no date cutoff**. When the HF mirror is
+stale (its "newest" rows are ~2-3 years old), users saw a map full of
+2022-2023 sightings labeled as the "last 60 days" layer.
+
+These tests pin the new behavior:
+
+* Rows older than ``_NUFORC_RECENT_DAYS`` are dropped before the take-top-N.
+* If the HF mirror has nothing in the window, the fallback returns ``[]``
+  and logs ERROR (don't silently serve stale data).
+* ``fetch_uap_sightings`` records the failure when BOTH paths fail, so
+  the layer shows as broken in the health registry instead of "fresh".
+"""
+
+from __future__ import annotations
+
+import logging
+from datetime import datetime as real_datetime
+
+
+class _FixedDateTime(real_datetime):
+    """A datetime whose utcnow() returns a pinned value, for deterministic
+    cutoff math. Subclasses real datetime so existing operations still work."""
+
+    @classmethod
+    def utcnow(cls):
+        return cls(2026, 5, 1, 12, 0, 0)
+
+
+class _StubResponse:
+    status_code = 200
+
+    def __init__(self, text: str):
+        self.text = text
+
+
+def _stub_geocode_cache(*_args, **_kwargs):
+    """Pre-populated location cache so the fallback doesn't try to hit
+    Photon during the test."""
+    return {
+        "Denver, CO, USA": [39.7392, -104.9903],
+        "Seattle, WA, USA": [47.6062, -122.3321],
+        "Phoenix, AZ, USA": [33.4484, -112.0740],
+    }
+
+
+def test_hf_fallback_drops_rows_older_than_60_days(monkeypatch):
+    """Pre-fix: a row from 2023 would make it into the layer if it was
+    among the newest 250 in the HF mirror. Post-fix: it's filtered out
+    before we even count to 250."""
+    from services.fetchers import earth_observation as eo
+
+    # 2026-05-01 - 60 days = 2026-03-02. So 2026-03-01 is one day too old.
+    csv_text = (
+        "Sighting,Occurred,Location,Shape,Duration,Posted,Summary\n"
+        '1,2026-04-15 21:00:00 Local,"Denver, CO, USA",Triangle,5 minutes,2026-04-16,"In-window sighting"\n'
+        '2,2023-06-01 21:00:00 Local,"Seattle, WA, USA",Light,30 seconds,2023-06-02,"Three years old"\n'
+        '3,2022-01-15 20:00:00 Local,"Phoenix, AZ, USA",Disk,2 minutes,2022-01-16,"Even older"\n'
+    )
+
+    monkeypatch.setattr(eo, "datetime", _FixedDateTime)
+    monkeypatch.setattr(eo, "fetch_with_curl", lambda *a, **kw: _StubResponse(csv_text))
+    monkeypatch.setattr(eo, "_load_nuforc_location_cache", _stub_geocode_cache)
+    monkeypatch.setattr(eo, "_save_nuforc_location_cache", lambda cache: None)
+    # If the cutoff is missing, the geocoder may still get called for the
+    # 2022/2023 rows. We assert geocoder is NEVER invoked for stale rows.
+    geocode_calls: list[str] = []
+
+    def _geocode_spy(location, city, state, country=""):
+        geocode_calls.append(location)
+        return None  # already in cache, shouldn't be hit anyway
+
+    monkeypatch.setattr(eo, "_geocode_uap_location", _geocode_spy)
+
+    sightings = eo._build_uap_sightings_from_hf_mirror()
+
+    ids = [s["id"] for s in sightings]
+    assert ids == ["NUFORC-1"], f"only the 2026 row should survive: got {ids}"
+    # Stale rows must not have been geocoded — they should be dropped
+    # before the geocoding loop is reached.
+    assert geocode_calls == []
+
+
+def test_hf_fallback_returns_empty_when_mirror_is_fully_stale(monkeypatch, caplog):
+    """The smoking-gun case: the HF mirror is so stale that NO rows are
+    within the rolling window. Pre-fix returned 250 ancient rows. Post-fix
+    returns ``[]`` and logs ERROR so the operator knows the layer is dead."""
+    from services.fetchers import earth_observation as eo
+
+    csv_text = (
+        "Sighting,Occurred,Location,Shape,Duration,Posted,Summary\n"
+        '1,2023-04-15 21:00:00 Local,"Denver, CO, USA",Triangle,5 minutes,2023-04-16,"Old"\n'
+        '2,2022-06-01 21:00:00 Local,"Seattle, WA, USA",Light,30 seconds,2022-06-02,"Older"\n'
+        '3,2021-01-15 20:00:00 Local,"Phoenix, AZ, USA",Disk,2 minutes,2021-01-16,"Ancient"\n'
+    )
+
+    monkeypatch.setattr(eo, "datetime", _FixedDateTime)
+    monkeypatch.setattr(eo, "fetch_with_curl", lambda *a, **kw: _StubResponse(csv_text))
+    monkeypatch.setattr(eo, "_load_nuforc_location_cache", _stub_geocode_cache)
+    monkeypatch.setattr(eo, "_save_nuforc_location_cache", lambda cache: None)
+    monkeypatch.setattr(eo, "_geocode_uap_location", lambda *a, **kw: None)
+
+    with caplog.at_level(logging.ERROR, logger="services.fetchers.earth_observation"):
+        sightings = eo._build_uap_sightings_from_hf_mirror()
+
+    assert sightings == []
+    # The error log should mention how many stale rows were dropped so the
+    # operator can tell the mirror is the problem (not "we got 0 rows" which
+    # could also mean the download failed).
+    relevant = [r for r in caplog.records if "HF fallback yielded 0 rows" in r.getMessage()]
+    assert relevant, "expected loud ERROR when HF mirror is fully stale"
+    # The message should report the count of dropped stale rows.
+    assert any("dropped 3" in r.getMessage() for r in relevant)
+
+
+def test_hf_fallback_still_returns_data_when_some_rows_are_in_window(monkeypatch):
+    """Mixed-age mirror: some rows in the window, some not. The fallback
+    should return only the in-window rows and not log the doomsday ERROR."""
+    from services.fetchers import earth_observation as eo
+
+    csv_text = (
+        "Sighting,Occurred,Location,Shape,Duration,Posted,Summary\n"
+        '1,2026-04-15 21:00:00 Local,"Denver, CO, USA",Triangle,5 minutes,2026-04-16,"Fresh"\n'
+        '2,2026-04-10 21:00:00 Local,"Seattle, WA, USA",Light,30 seconds,2026-04-10,"Also fresh"\n'
+        '3,2020-01-15 20:00:00 Local,"Phoenix, AZ, USA",Disk,2 minutes,2020-01-16,"Ancient"\n'
+    )
+
+    monkeypatch.setattr(eo, "datetime", _FixedDateTime)
+    monkeypatch.setattr(eo, "fetch_with_curl", lambda *a, **kw: _StubResponse(csv_text))
+    monkeypatch.setattr(eo, "_load_nuforc_location_cache", _stub_geocode_cache)
+    monkeypatch.setattr(eo, "_save_nuforc_location_cache", lambda cache: None)
+    monkeypatch.setattr(eo, "_geocode_uap_location", lambda *a, **kw: None)
+
+    sightings = eo._build_uap_sightings_from_hf_mirror()
+
+    ids = sorted(s["id"] for s in sightings)
+    assert ids == ["NUFORC-1", "NUFORC-2"], f"only in-window rows should appear: got {ids}"
+
+
+def test_fetch_uap_sightings_marks_failure_when_both_paths_empty(monkeypatch, caplog):
+    """When the live path raises AND the HF fallback returns empty,
+    ``fetch_uap_sightings`` must:
+      * NOT mark the layer fresh (pre-fix bug: it did, so the layer
+        showed as healthy-but-empty for days)
+      * call ``assert_canary("uap_sightings", 0)`` so the health
+        registry surfaces the broken layer
+      * log an ERROR with the live-path exception for debugging
+    """
+    from services.fetchers import earth_observation as eo
+    from services.fetchers import _store
+
+    monkeypatch.setattr(_store, "is_any_active", lambda layer: True)
+    monkeypatch.setattr(eo, "_load_nuforc_sightings_cache", lambda force_refresh=False: None)
+
+    def _boom():
+        raise RuntimeError("NUFORC live: zero rows pulled across 3 months")
+
+    monkeypatch.setattr(eo, "_build_recent_uap_sightings", _boom)
+    monkeypatch.setattr(eo, "_build_uap_sightings_from_hf_mirror", lambda: [])
+
+    marked: list[str] = []
+    monkeypatch.setattr(eo, "_mark_fresh", lambda *keys: marked.extend(keys))
+
+    canary_calls: list[tuple[str, int]] = []
+    import services.slo as slo
+    monkeypatch.setattr(
+        slo, "assert_canary", lambda key, value: canary_calls.append((key, int(value)))
+    )
+
+    with caplog.at_level(logging.ERROR, logger="services.fetchers.earth_observation"):
+        eo.fetch_uap_sightings()
+
+    assert marked == [], "broken layer must NOT be marked fresh"
+    assert canary_calls == [("uap_sightings", 0)], (
+        f"expected canary trip when both paths fail; got {canary_calls}"
+    )
+    # The live error message should propagate into the error log so the
+    # operator can tell live failed AND fallback was empty (not the other
+    # way around).
+    assert any(
+        "both live NUFORC and HF fallback" in r.getMessage()
+        for r in caplog.records
+    )
+
+
+def test_fetch_uap_sightings_succeeds_when_fallback_returns_data(monkeypatch):
+    """Positive path: live fails, fallback returns rows. The layer is
+    populated and marked fresh; assert_canary is NOT tripped (we only
+    trip the canary when the layer has zero data)."""
+    from services.fetchers import earth_observation as eo
+    from services.fetchers import _store
+
+    monkeypatch.setattr(_store, "is_any_active", lambda layer: True)
+    monkeypatch.setattr(eo, "_load_nuforc_sightings_cache", lambda force_refresh=False: None)
+    monkeypatch.setattr(
+        eo, "_build_recent_uap_sightings", lambda: (_ for _ in ()).throw(RuntimeError("live down"))
+    )
+
+    fallback_rows = [{"id": "NUFORC-fb-1", "date_time": "2026-04-20", "lat": 0.0, "lng": 0.0}]
+    monkeypatch.setattr(eo, "_build_uap_sightings_from_hf_mirror", lambda: fallback_rows)
+    monkeypatch.setattr(eo, "_save_nuforc_sightings_cache", lambda s: None)
+
+    marked: list[str] = []
+    monkeypatch.setattr(eo, "_mark_fresh", lambda *keys: marked.extend(keys))
+
+    canary_calls: list[tuple[str, int]] = []
+    import services.slo as slo
+    monkeypatch.setattr(
+        slo, "assert_canary", lambda key, value: canary_calls.append((key, int(value)))
+    )
+
+    eo.fetch_uap_sightings()
+
+    assert marked == ["uap_sightings"]
+    assert canary_calls == [], "canary should not trip when fallback supplies data"
+
+
+def test_uap_scheduler_runs_weekly_not_daily():
+    """The cron job for the UAP layer must be configured for Mondays at
+    12:00 UTC, not daily. Daily was the pre-fix default; weekly matches
+    the layer's stated cadence (a rolling 60-day digest) and keeps load
+    on nuforc.org light."""
+    from services import data_fetcher
+
+    src = data_fetcher.__file__
+    with open(src, "r", encoding="utf-8") as f:
+        text = f.read()
+
+    # Anchor on the scheduler block by id, then assert the cron triggers.
+    assert "uap_sightings_weekly" in text, (
+        "scheduler id should be uap_sightings_weekly (was uap_sightings_daily pre-fix)"
+    )
+    # The day_of_week directive is the difference between daily and weekly.
+    # If somebody flips it back to daily, this fires.
+    weekly_block = text.split("uap_sightings_weekly", 1)[0]
+    # Walk backwards for the matching add_job call.
+    add_job_idx = weekly_block.rfind("add_job(")
+    assert add_job_idx >= 0, "could not locate add_job block for UAP scheduler"
+    job_block = text[add_job_idx : text.find(")", text.index("uap_sightings_weekly")) + 1]
+    assert 'day_of_week="mon"' in job_block, (
+        f"expected day_of_week='mon' in UAP scheduler block:\n{job_block}"
+    )

desktop-shell/package-lock.json

@@ -1,12 +1,12 @@
 {
   "name": "@shadowbroker/desktop-shell",
-  "version": "0.9.79",
+  "version": "0.9.81",
   "lockfileVersion": 3,
   "requires": true,
   "packages": {
     "": {
       "name": "@shadowbroker/desktop-shell",
-      "version": "0.9.79",
+      "version": "0.9.81",
       "devDependencies": {
         "typescript": "^5.6.0"
       }

desktop-shell/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@shadowbroker/desktop-shell",
-  "version": "0.9.79",
+  "version": "0.9.81",
   "private": true,
   "description": "ShadowBroker desktop shell packaging, runtime bridge, and release tooling",
   "scripts": {

desktop-shell/tauri-skeleton/scripts/build-backend-runtime.cjs

@@ -130,6 +130,45 @@ function stageBackendRuntime() {
   });
   stagePrivacyCoreArtifact();
   stageReleaseAttestation();
+  stageStartScripts();
+}
+
+/**
+ * Copy ``start.bat`` and ``start.sh`` from the repo root into the
+ * staged backend-runtime/ so they sit next to ``privacy_core.dll``.
+ *
+ * Why: an MSI/EXE/AppImage user who wants to launch via the dev-style
+ * scripts (because the desktop shell is failing, or they prefer the
+ * browser frontend at localhost:3000) shouldn't have to clone the
+ * source repo just to get the scripts. Having them inside the install
+ * directory also means the bundled ``privacy_core.dll`` fallback in
+ * those scripts resolves to the SAME directory as the script, which
+ * is exactly the layout the v0.9.81 script update is looking for.
+ *
+ * Tracked from issue #319: users who fell back to start.bat from
+ * their MSI install dir had to go fetch it from GitHub, then saw a
+ * scary "install Rust" warning because the script didn't know where
+ * the bundled DLL was. Bundling the script removes both problems.
+ */
+function stageStartScripts() {
+  const scripts = ['start.bat', 'start.sh'];
+  for (const name of scripts) {
+    const src = path.join(repoRoot, name);
+    if (!fs.existsSync(src)) {
+      console.warn(`backend-runtime staged without ${name} (not at repo root)`);
+      continue;
+    }
+    const dst = path.join(outputDir, name);
+    fs.copyFileSync(src, dst);
+    // Preserve executable bit on POSIX systems for the .sh script.
+    if (name.endsWith('.sh') && process.platform !== 'win32') {
+      try {
+        fs.chmodSync(dst, 0o755);
+      } catch {
+        /* best-effort; not fatal on filesystems that don't honor chmod */
+      }
+    }
+  }
 }
 
 function stagePrivacyCoreArtifact() {

desktop-shell/tauri-skeleton/scripts/build-frontend-export.cjs

@@ -46,12 +46,18 @@ function prepareBuildTree() {
   const stagedLayoutPath = path.join(buildFrontendDir, 'src', 'app', 'layout.tsx');
   if (fs.existsSync(stagedLayoutPath)) {
     const layoutSource = fs.readFileSync(stagedLayoutPath, 'utf8');
+    // CRLF compatibility: on Windows checkouts without ``core.autocrlf=input``
+    // (the default) layout.tsx has CRLF line endings, but the original regexes
+    // only matched LF. The strip silently no-op'd, ``force-dynamic`` stayed,
+    // and Next's static-export refused to render ``/_not-found`` ("Page with
+    // `dynamic = \"force-dynamic\"` couldn't be exported"). Use ``\r?\n`` so
+    // the strip works regardless of line-ending normalization.
     fs.writeFileSync(
       stagedLayoutPath,
       layoutSource
-        .replace(/\n\/\/ The dashboard is a live local runtime[\s\S]*?client polling ever hydrates\.\n/g, '\n')
-        .replace(/\nexport const dynamic = ['"]force-dynamic['"];\n/g, '\n')
-        .replace(/\nexport const revalidate = 0;\n/g, '\n'),
+        .replace(/\r?\n\/\/ The dashboard is a live local runtime[\s\S]*?client polling ever hydrates\.\r?\n/g, '\n')
+        .replace(/\r?\nexport const dynamic = ['"]force-dynamic['"];\r?\n/g, '\n')
+        .replace(/\r?\nexport const revalidate = 0;\r?\n/g, '\n'),
     );
   }
 

desktop-shell/tauri-skeleton/src-tauri/Cargo.lock

@@ -4201,7 +4201,7 @@ dependencies = [
 
 [[package]]
 name = "shadowbroker-tauri-shell"
-version = "0.9.79"
+version = "0.9.81"
 dependencies = [
  "axum",
  "base64 0.22.1",

desktop-shell/tauri-skeleton/src-tauri/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "shadowbroker-tauri-shell"
-version = "0.9.79"
+version = "0.9.81"
 edition = "2021"
 
 [build-dependencies]

desktop-shell/tauri-skeleton/src-tauri/tauri.conf.json

@@ -1,7 +1,7 @@
 {
   "$schema": "https://schema.tauri.app/config/2",
   "productName": "ShadowBroker",
-  "version": "0.9.79",
+  "version": "0.9.81",
   "identifier": "com.shadowbroker.desktop",
   "build": {
     "frontendDist": "../../../frontend/out",
@@ -38,7 +38,7 @@
   },
   "plugins": {
     "updater": {
-      "pubkey": "dW50cnVzdGVkIGNvbW1lbnQ6IG1pbmlzaWduIHB1YmxpYyBrZXk6IEUxODExMjQ4MkJBMThFNTgKUldSWWpxRXJTQktCNFF3ZXNQbndUK0pVWUEwNDNuajcrUGI3ZEI4TWtDUDlQdHhudmlHUkNjQUUK",
+      "pubkey": "dW50cnVzdGVkIGNvbW1lbnQ6IG1pbmlzaWduIHB1YmxpYyBrZXk6IDVEMTFERDdCNjhBRTk3MDcKUldRSGw2NW9lOTBSWGRjS1ZobFN5TkZsd3NkZ2g2L09WZzU4aytTR2FtN3ZtR0ZKejlNNldTbFUK",
       "endpoints": [
         "https://github.com/BigBodyCobain/Shadowbroker/releases/latest/download/latest.json"
       ],

docker-compose.relay.yml

@@ -6,7 +6,7 @@ services:
       dockerfile: ./backend/Dockerfile
     container_name: shadowbroker-relay
     ports:
-      - "0.0.0.0:8000:8000"
+      - "127.0.0.1:8000:8000"
     env_file: .env
     volumes:
       - relay_data:/app/data

docker-compose.yml

@@ -24,8 +24,16 @@ services:
       # Private Infonet bootstrap seeds. Seeds are discovery hints, not fixed roots.
       - MESH_BOOTSTRAP_SEED_PEERS=${MESH_BOOTSTRAP_SEED_PEERS:-http://gqpbunqbgtkcqilvclm3xrkt3zowjyl3s62kkktvojgvxzizamvbrqid.onion:8000}
       - MESH_DEFAULT_SYNC_PEERS=${MESH_DEFAULT_SYNC_PEERS:-}
+      - MESH_SYNC_TIMEOUT_S=${MESH_SYNC_TIMEOUT_S:-5}
+      - MESH_RELAY_PUSH_TIMEOUT_S=${MESH_RELAY_PUSH_TIMEOUT_S:-45}
+      # Explicitly opt into HTTPS/IP-based peer sync. Default remains private transports only.
+      - MESH_INFONET_ALLOW_CLEARNET_SYNC=${MESH_INFONET_ALLOW_CLEARNET_SYNC:-false}
+      # Tor/Arti SOCKS transport for private .onion Infonet sync.
+      - MESH_ARTI_ENABLED=${MESH_ARTI_ENABLED:-false}
+      - MESH_ARTI_SOCKS_PORT=${MESH_ARTI_SOCKS_PORT:-9050}
       # Operator-trusted sync/push peers. Leave empty unless you control the peer secret on both sides.
       - MESH_RELAY_PEERS=${MESH_RELAY_PEERS:-}
+      - MESH_PUBLIC_PEER_URL=${MESH_PUBLIC_PEER_URL:-}
       # Shared transport auth for operator peer push. Must be set to a unique secret per deployment.
       - MESH_PEER_PUSH_SECRET=${MESH_PEER_PUSH_SECRET:-}
       # Issue #256: optional per-peer HMAC secrets. Comma-separated

frontend/package-lock.json

@@ -1,12 +1,12 @@
 {
   "name": "frontend",
-  "version": "0.9.79",
+  "version": "0.9.81",
   "lockfileVersion": 3,
   "requires": true,
   "packages": {
     "": {
       "name": "frontend",
-      "version": "0.9.79",
+      "version": "0.9.81",
       "dependencies": {
         "@mapbox/point-geometry": "^1.1.0",
         "@tauri-apps/plugin-process": "^2.3.1",

frontend/package.json

@@ -1,6 +1,6 @@
 {
   "name": "frontend",
-  "version": "0.9.79",
+  "version": "0.9.81",
   "private": true,
   "scripts": {
     "dev": "node scripts/dev-all.cjs",

frontend/src/__tests__/desktop/updateRuntime.test.ts

@@ -9,12 +9,12 @@ import {
 } from '@/lib/updateRuntime';
 
 const RELEASE: GitHubLatestRelease = {
-  html_url: 'https://github.com/BigBodyCobain/Shadowbroker/releases/tag/v0.9.79',
+  html_url: 'https://github.com/BigBodyCobain/Shadowbroker/releases/tag/v0.9.81',
   assets: [
-    { name: 'ShadowBroker_0.9.79_x64_en-US.msi', browser_download_url: 'https://example.test/windows.msi' },
-    { name: 'ShadowBroker_0.9.79_x64-setup.exe', browser_download_url: 'https://example.test/windows-setup.exe' },
-    { name: 'ShadowBroker_0.9.79_aarch64.dmg', browser_download_url: 'https://example.test/macos.dmg' },
-    { name: 'ShadowBroker_0.9.79_amd64.AppImage', browser_download_url: 'https://example.test/linux.AppImage' },
+    { name: 'ShadowBroker_0.9.81_x64_en-US.msi', browser_download_url: 'https://example.test/windows.msi' },
+    { name: 'ShadowBroker_0.9.81_x64-setup.exe', browser_download_url: 'https://example.test/windows-setup.exe' },
+    { name: 'ShadowBroker_0.9.81_aarch64.dmg', browser_download_url: 'https://example.test/macos.dmg' },
+    { name: 'ShadowBroker_0.9.81_amd64.AppImage', browser_download_url: 'https://example.test/linux.AppImage' },
   ],
 };
 

frontend/src/__tests__/mesh/infonetShellGateDirectory.test.tsx

@@ -0,0 +1,57 @@
+import React from 'react';
+import { act, cleanup, render, screen } from '@testing-library/react';
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
+
+vi.mock('@/mesh/meshIdentity', () => ({
+  getNodeIdentity: vi.fn(() => ({ publicKey: 'test-public-key', nodeId: '!sb_test' })),
+  getWormholeIdentityDescriptor: vi.fn(() => ({ publicKey: 'wormhole-public-key' })),
+}));
+
+vi.mock('@/mesh/wormholeIdentityClient', () => ({
+  activateWormholeGatePersona: vi.fn(),
+  createWormholeGatePersona: vi.fn(),
+  enterWormholeGate: vi.fn(),
+  fetchWormholeIdentity: vi.fn(),
+  listWormholeGatePersonas: vi.fn(async () => ({ personas: [] })),
+}));
+
+vi.mock('@/components/InfonetTerminal/GateView', () => ({ default: () => <div>Gate view</div> }));
+vi.mock('@/components/InfonetTerminal/MarketView', () => ({ default: () => <div>Market view</div> }));
+vi.mock('@/components/InfonetTerminal/ProfileView', () => ({ default: () => <div>Profile view</div> }));
+vi.mock('@/components/InfonetTerminal/MessagesView', () => ({ default: () => <div>Messages view</div> }));
+vi.mock('@/components/InfonetTerminal/TerminalDashboard', () => ({ default: () => <div>Dashboard</div> }));
+vi.mock('@/components/InfonetTerminal/WeatherWidget', () => ({ default: () => <div>Weather</div> }));
+vi.mock('@/components/InfonetTerminal/TrendingPosts', () => ({ default: () => <div>Trending</div> }));
+vi.mock('@/components/InfonetTerminal/HashchainEvents', () => ({ default: () => <div>Hashchain</div> }));
+vi.mock('@/components/InfonetTerminal/NetworkStats', () => ({ default: () => <div>Network stats</div> }));
+vi.mock('@/components/InfonetTerminal/AIQueryView', () => ({ default: () => <div>AI view</div> }));
+vi.mock('@/components/InfonetTerminal/PetitionsView', () => ({ default: () => <div>Petitions</div> }));
+vi.mock('@/components/InfonetTerminal/UpgradeView', () => ({ default: () => <div>Upgrades</div> }));
+vi.mock('@/components/InfonetTerminal/ResolutionView', () => ({ default: () => <div>Resolution</div> }));
+vi.mock('@/components/InfonetTerminal/GateShutdownView', () => ({ default: () => <div>Gate shutdown</div> }));
+vi.mock('@/components/InfonetTerminal/BootstrapView', () => ({ default: () => <div>Bootstrap</div> }));
+vi.mock('@/components/InfonetTerminal/FunctionKeyView', () => ({ default: () => <div>Function keys</div> }));
+
+describe('InfonetShell gate directory', () => {
+  beforeEach(() => {
+    vi.useFakeTimers();
+  });
+
+  afterEach(() => {
+    vi.useRealTimers();
+    cleanup();
+  });
+
+  it('renders available gates under the landing logo section', async () => {
+    const { default: InfonetShell } = await import('@/components/InfonetTerminal/InfonetShell');
+
+    render(<InfonetShell isOpen onClose={() => {}} />);
+    await act(async () => {
+      vi.advanceTimersByTime(2500);
+    });
+
+    expect(screen.getByText('AVAILABLE OBFUSCATED GATES:')).toBeTruthy();
+    expect(screen.getByRole('button', { name: /\[>\]\s*infonet/i })).toBeTruthy();
+    expect(screen.getByRole('button', { name: /\[>\]\s*gathered-intel/i })).toBeTruthy();
+  });
+});

frontend/src/app/page.tsx

@@ -39,6 +39,7 @@ import { useFeedHealth } from '@/hooks/useFeedHealth';
 import { useKeyboardShortcuts } from '@/hooks/useKeyboardShortcuts';
 import KeyboardShortcutsOverlay from '@/components/KeyboardShortcutsOverlay';
 import AlertToast from '@/components/AlertToast';
+import AisUpstreamBanner from '@/components/AisUpstreamBanner';
 import { useAlertToasts } from '@/hooks/useAlertToasts';
 import { useWatchlist } from '@/hooks/useWatchlist';
 import WatchlistWidget from '@/components/WatchlistWidget';
@@ -933,6 +934,11 @@ export default function Dashboard() {
           onFlyTo={handleFlyTo}
         />
 
+        {/* AIS UPSTREAM OUTAGE BANNER — renders only when AIS is configured
+            but the WebSocket upstream is unreachable. Tells users the empty
+            ocean isn't their fault. */}
+        <AisUpstreamBanner />
+
         {/* ONBOARDING MODAL */}
         {showOnboarding && (
           <OnboardingModal

frontend/src/components/AisUpstreamBanner.tsx

@@ -0,0 +1,61 @@
+/**
+ * AisUpstreamBanner — visible notice that AIS ship data is unavailable
+ * because the upstream provider (AISStream) is offline.
+ *
+ * Renders nothing when AIS is healthy or when AIS isn't configured at all.
+ * Mounted at the app shell level so users see it before they wonder why
+ * the ocean looks empty.
+ */
+import { useState } from 'react';
+import { useAisUpstreamHealth } from '@/hooks/useAisUpstreamHealth';
+
+export function AisUpstreamBanner() {
+  const health = useAisUpstreamHealth();
+  const [dismissed, setDismissed] = useState(false);
+
+  if (!health || !health.aisEnabled || health.connected || dismissed) {
+    return null;
+  }
+
+  // Format the staleness for the operator. ``null`` means we never received
+  // anything since startup; otherwise show minutes if > 60s.
+  let stalenessLabel = 'never received';
+  if (health.lastMsgAgeSeconds != null) {
+    const minutes = Math.floor(health.lastMsgAgeSeconds / 60);
+    if (minutes >= 1) {
+      stalenessLabel = `last update ${minutes} min ago`;
+    } else {
+      stalenessLabel = `last update ${health.lastMsgAgeSeconds}s ago`;
+    }
+  }
+
+  return (
+    <div
+      role="status"
+      aria-live="polite"
+      className="pointer-events-auto fixed top-3 left-1/2 z-[100] -translate-x-1/2 max-w-[640px] rounded-md border border-amber-500/60 bg-amber-900/85 px-4 py-2 text-sm text-amber-50 shadow-lg backdrop-blur"
+    >
+      <div className="flex items-start gap-3">
+        <span aria-hidden className="mt-0.5 text-amber-300">⚠</span>
+        <div className="flex-1">
+          <div className="font-semibold">Ship data temporarily unavailable</div>
+          <div className="text-xs opacity-90">
+            AISStream upstream is offline ({stalenessLabel}). The map will
+            refill once their service comes back online — nothing is wrong
+            with your install.
+          </div>
+        </div>
+        <button
+          type="button"
+          onClick={() => setDismissed(true)}
+          aria-label="Dismiss"
+          className="text-amber-200 hover:text-white"
+        >
+          ✕
+        </button>
+      </div>
+    </div>
+  );
+}
+
+export default AisUpstreamBanner;

frontend/src/components/ChangelogModal.tsx

@@ -20,129 +20,82 @@ import {
   Heart,
 } from 'lucide-react';
 
-const CURRENT_VERSION = '0.9.79';
+const CURRENT_VERSION = '0.9.81';
 const STORAGE_KEY = `shadowbroker_changelog_v${CURRENT_VERSION}`;
-const RELEASE_TITLE = 'Onboarding, Live Feeds, Mesh, and Agent Hardening';
+const RELEASE_TITLE = 'Signed Auto-Update + Update Button Race Fix';
 
 const HEADLINE_FEATURES = [
   {
-    icon: <Bot size={20} className="text-purple-400" />,
+    icon: <KeyRound size={20} className="text-purple-400" />,
     accent: 'purple' as const,
-    title: 'Agentic onboarding for OpenClaw-compatible agents',
-    subtitle: 'First-time setup now includes local/direct agent connection, access-tier selection, copyable HMAC setup, and optional Tor hidden-service prep.',
+    title: 'Signed Auto-Update Going Forward (one manual hop)',
+    subtitle: 'After installing v0.9.81, the in-app Update button finally works end-to-end. This release establishes a fresh signing key — every release from here is a one-click upgrade.',
     details: [
-      'The onboarding flow can generate the local agent connection bundle through the existing HMAC API, point agents at /api/ai/tools, and let operators choose restricted read-only or full write access before connecting an agent.',
-      'Remote mode is labeled honestly: .onion exposes the signed HTTP agent API over Tor. Wormhole/MLS is not claimed as the current agent command transport.',
-      'The setup copy works for OpenClaw, Hermes, or any custom agent that implements the documented HMAC request contract.',
+      'tauri.conf.json now carries a fresh minisign pubkey (the previous keypair was generated before v0.9.79 shipped but the matching private key was lost before any release was actually signed, so no release before v0.9.81 has working auto-update).',
+      'The v0.9.81 release artifacts ship with a signed latest.json + .sig files so every install on v0.9.81 or later can verify and apply the next release automatically via the Tauri updater plugin.',
+      'One-time cost: if you are upgrading from v0.9.79 or v0.9.8, the click-Update path falls back to a manual download because the new pubkey does not match the one baked into your install. Click the MANUAL DOWNLOAD button in the update dialog → grab the .msi from the release page → run it → from then on auto-update works in-app.',
     ],
-    callToAction: 'OPEN FIRST-TIME SETUP -> AI AGENT',
+    callToAction: 'CLICK UPDATE → DOWNLOAD MSI ONCE → AUTO-UPDATE FOREVER',
   },
   {
-    icon: <Bot size={20} className="text-purple-400" />,
-    accent: 'purple' as const,
-    title: 'Agentic AI Channel — supports OpenClaw and any HMAC-signing agent',
-    subtitle: 'ShadowBroker now exposes a signed agent command channel. Bring your own agent (OpenClaw, Claude Code, GPT, LangChain, or a custom client) and drive the dashboard from any LLM that speaks the protocol.',
-    details: [
-      'A signed command channel (POST /api/ai/channel/command) plus a batched concurrent-execution endpoint (up to 20 tool calls per round-trip via /api/ai/channel/batch). Agents query flights, ships, SIGINT, news, and intel layers; reason over the live mesh; and run market or threat analyses without a human in the loop.',
-      'HMAC-SHA256 request signing with timestamp + nonce replay protection. Tier-gated access (restricted vs full) governs which read and write commands the agent can invoke. Every call is auditable through the channel log.',
-      'ShadowBroker does not bundle an LLM, an agent runtime, or model weights — it ships the protocol. Any agent that signs requests with the documented HMAC contract can connect. OpenClaw is the reference implementation.',
-    ],
-    callToAction: 'CONNECT YOUR AGENT \u2192 /API/AI/CHANNEL/COMMAND',
-  },
-  {
-    icon: <Network size={20} className="text-cyan-400" />,
+    icon: <Network size={20} className="text-amber-400" />,
     accent: 'cyan' as const,
-    title: 'InfoNet Testnet \u2014 Framework, Privacy, and a Path to Decentralized Intelligence',
-    subtitle: 'The testnet now ships its full governance economy and the runway for a privacy-preserving decentralized intelligence platform.',
+    title: 'AIS Maritime Resilience — Outage Banner + AISHub Fallback',
+    subtitle: 'When AISStream’s WebSocket goes offline (as happened upstream in May 2026), the ships layer no longer goes silently empty.',
     details: [
-      'Sovereign Shell views: petitions (governance DSL covers parameter updates and feature toggles), upgrade-hash voting (80% supermajority, 67% Heavy-Node activation), evidence submission, dispute markets, gate suspension and shutdown, and bootstrap eligible-node-one-vote. Every write action is a clickable form with verbatim diagnostics on rejection.',
-      'Privacy primitive runway: locked Protocol contracts for ring signatures, stealth addresses, shielded balances, and DEX matching. The privacy-core Rust crate is the integration target. Function Keys (anonymous citizenship proof) ship 5 of 6 pieces; only blind-signature issuance waits on a primitive decision.',
-      'Backbone: two-tier event state with epoch finality, identity rotation, progressive penalties, ramp milestones, and constitutional invariants enforced via MappingProxyType. Sprint 11+ wires the cryptographic primitives into the locked Protocols.',
-      'Still an experimental testnet \u2014 no privacy guarantee yet. Treat all channels as public until E2E and the privacy primitives ship.',
+      'AIS proxy health surfaces in /api/health: connected, last_msg_age_seconds, proxy_spawn_count. A dismissible amber banner explains the outage (“Ship data temporarily unavailable — AISStream upstream is offline”) instead of letting users assume their install is broken.',
+      'AISHub REST fallback (free tier at aishub.net/api). Polls every 20 minutes when the primary is disconnected and merges vessels into the same store with source: “aishub” so existing tooling attributes the provider.',
+      'Live data wins races: if the WebSocket reconnects mid-poll, fresh AISStream updates aren’t overwritten by stale REST records. Opt-in via AISHUB_USERNAME; cadence configurable via AISHUB_POLL_INTERVAL_MINUTES (clamped [1, 360]).',
     ],
-    callToAction: 'OPEN SOVEREIGN SHELL \u2192 PETITIONS \u2022 UPGRADES \u2022 GATES',
+    callToAction: 'SET AISHUB_USERNAME \u2192 RESTART BACKEND',
+  },
+  {
+    icon: <Shield size={20} className="text-cyan-400" />,
+    accent: 'cyan' as const,
+    title: 'Data-Layer Repair \u2014 UAP Cutoff + GPS Jamming Detection',
+    subtitle: 'Two long-broken layers fixed at the source. UFO sightings are actually recent now; GPS jamming zones actually fire.',
+    details: [
+      'UAP sightings: the Hugging Face NUFORC mirror fallback had no date cutoff, so when the live nuforc.org scrape failed the layer served 3-year-old reports as \u201crecent\u201d. Now drops rows older than 60 days and logs loudly when the mirror is fully stale. Scheduler moved daily \u2192 weekly (Mondays 12:00 UTC).',
+      'GPS jamming: three stacked filters meant the layer almost never lit up. nac_p == 0 (\u201cGPS lock lost\u201d) was filtered out as if it were an old transponder \u2014 it\u2019s actually the strongest jamming signal. Now counted. MIN_AIRCRAFT lowered 5 \u2192 3 so sparser hotspots clear; MIN_RATIO lowered 0.30 \u2192 0.20.',
+      'Both layers now surface their own outages via assert_canary so operators see broken vs empty, not silently stale.',
+    ],
+    callToAction: 'TOGGLE UAP \u2022 GPS JAMMING LAYERS',
   },
 ];
 
 const NEW_FEATURES = [
   {
-    icon: <Clock size={18} className="text-cyan-400" />,
-    title: 'Startup and Feed Responsiveness Pass',
-    desc: 'Map-critical feeds now lean on startup caches and priority preload behavior so the dashboard can paint before heavyweight synthesis jobs finish.',
-  },
-  {
-    icon: <Network size={18} className="text-green-400" />,
-    title: 'MeshChat MQTT Settings',
-    desc: 'Public MeshChat stays opt-in and now has an in-panel settings lane for broker, port, username, password, and channel PSK while remaining separated from Wormhole/private mode.',
+    icon: <Plane size={18} className="text-orange-400" />,
+    title: 'Cumulative Fuel & CO2 per Flight',
+    desc: 'Aircraft tooltip now shows how much fuel each plane has actually burned in the air since first observation, not just the per-hour rate. 15-minute gap between sightings resets the session; 24-hour clamp protects against clock skew; per-icao prune every 5 minutes keeps memory bounded.',
   },
   {
     icon: <Plane size={18} className="text-cyan-400" />,
-    title: 'Selected Entity Trails',
-    desc: 'Flight and vessel trails are drawn only for selected assets, reducing global clutter while still exposing movement history for unknown-route entities.',
+    title: 'Per-Flight Source Attribution',
+    desc: 'Every aircraft record now carries a source field (adsb.lol, OpenSky, airplanes.live, adsb.fi) so consumers can attribute the data provider. Pre-fix, adsb.lol records were unmarked while OpenSky records were explicitly tagged, making it look like adsb.lol was unused even though it is the primary source.',
   },
   {
-    icon: <Plane size={18} className="text-amber-400" />,
-    title: 'Aircraft Detail Cards',
-    desc: 'Commercial aircraft stay airline-first, while private and general aviation aircraft can show model-focused Wiki context and imagery when available.',
+    icon: <Network size={18} className="text-green-400" />,
+    title: 'Cross-Node DM Mailbox Replication',
+    desc: 'Direct messages now replicate across mesh nodes when one party is offline. Per-(sender, recipient) anti-spam cap enforced as a network rule (not client-side) so source-code tampering cannot bypass it.',
   },
   {
-    icon: <Cpu size={18} className="text-purple-400" />,
-    title: 'AI Batch Command Channel',
-    desc: 'POST up to 20 tool calls in a single HTTP round-trip; the backend executes them concurrently and returns a fan-out result map. Cuts agent latency by an order of magnitude over sequential calls.',
-  },
-  {
-    icon: <Scale size={18} className="text-amber-400" />,
-    title: 'Governance DSL — Petition-Driven Parameter Changes',
-    desc: 'Type-safe payload executor for UPDATE_PARAM, BATCH_UPDATE_PARAMS, ENABLE_FEATURE, and DISABLE_FEATURE petitions. Tunable knobs change on-chain via a vote — no code deploys required.',
-  },
-  {
-    icon: <GitBranch size={18} className="text-purple-400" />,
-    title: 'Upgrade-Hash Governance',
-    desc: 'Protocol upgrades that need new logic (not just parameter changes) vote on a SHA-256 hash of the verified release. 80% supermajority, 40% quorum, 67% Heavy-Node activation. Lifecycle: signatures, voting, challenge window, awaiting readiness, activated.',
-  },
-  {
-    icon: <KeyRound size={18} className="text-purple-400" />,
-    title: 'Function Keys — Anonymous Citizenship Proof',
-    desc: 'A citizen proves "I am an Infonet citizen" without revealing their Infonet identity. 5 of 6 pieces shipped: nullifiers, challenge-response, two-phase commit receipts, enumerated denial codes, batched settlement. Issuance via blind signatures waits on a primitive decision.',
-  },
-  {
-    icon: <Shield size={18} className="text-cyan-400" />,
-    title: 'Privacy Primitive Runway',
-    desc: 'Locked Protocol contracts in services/infonet/privacy/contracts.py for ring signatures, stealth addresses, Pedersen commitments, range proofs, and DEX matching. The privacy-core Rust crate is the integration target — no caller of the privacy module needs to know which scheme is active.',
-  },
-  {
-    icon: <Layers size={18} className="text-blue-400" />,
-    title: 'Two-Tier State + Epoch Finality',
-    desc: 'Tier 1 events propagate CRDT-style for low latency; Tier 2 events require epoch finality before they can be acted on. Identity rotation, progressive penalties, ramp milestones, and constitutional invariants are enforced via MappingProxyType.',
-  },
-  {
-    icon: <Terminal size={18} className="text-cyan-400" />,
-    title: 'Sovereign Shell Write Surface',
-    desc: 'PetitionsView, UpgradeView, ResolutionView, GateShutdownView, BootstrapView, and FunctionKeyView each expose every Sprint 4-8 + 10 write action as a clickable form. Adaptive polling tightens to 8 seconds during active voting/challenge phases.',
-  },
-  {
-    icon: <Clock size={18} className="text-pink-400" />,
-    title: 'Time Machine — Snapshot Playback',
-    desc: 'Scrub backward through saved telemetry. Live polling pauses on entry to snapshot mode, the map redraws from the recorded snapshot, and moving entities interpolate between recorded frames. Hourly index lets you jump to any captured timestamp; pressing Live restores the current feed instantly.',
-  },
-  {
-    icon: <Satellite size={18} className="text-orange-400" />,
-    title: 'SAR Satellite Telemetry — ASF, OPERA, Copernicus',
-    desc: 'New SAR (Synthetic Aperture Radar) layer. Mode A (default-on) pulls free catalog metadata from the Alaska Satellite Facility — no account required. Mode B (two-step opt-in) ingests pre-processed ground-change anomalies from NASA OPERA, Copernicus EGMS, GFM, EMS, and UNOSAT — deformation, flood, and damage assessments. Integrates with OpenClaw so agents can read and act on SAR anomalies; broadcasts default to private-tier transport (Tor / RNS).',
+    icon: <Clock size={18} className="text-amber-400" />,
+    title: 'Infonet Sync — HTTP 429 Honored',
+    desc: 'When an upstream peer returns Retry-After, the node now waits exactly that long instead of retrying every 60 seconds and keeping the upstream rate-limit bucket permanently full. Exponential backoff on consecutive failures capped at 30 minutes.',
   },
 ];
 
 const BUG_FIXES = [
-  'Docker proxy and backend port handling hardened so changing the host backend port does not require changing the internal service contract.',
-  'Global Threat Intercept and live-data startup paths no longer wait on slow-tier synthesis before cached data can paint the UI.',
-  'MeshChat and Infonet statuses now separate public MQTT participation, private Wormhole mode, and local node bootstrap so the UI does not imply the wrong connection state.',
-  'Commercial aircraft detail cards no longer show a confusing model image alongside the airline card.',
-  'Sovereign Shell adaptive polling — voting and challenge windows refresh every 8 seconds while active, every 30 to 60 seconds when idle. Voting feels live without a websocket layer.',
-  'Per-row write actions (petitions, upgrades, disputes) hold isolated submission state so concurrent forms no longer share a single in-flight slot.',
-  'Verbatim diagnostic surfacing on every write button. The backend reason text is always shown on rejection — no opaque "denied" toasts.',
-  'Evidence submission canonicalization matches Python repr() exactly, so client-side SHA-256 hashes round-trip cleanly through the chain.',
-  'Function Keys copy is context-agnostic — citizenship proof is described abstractly, not tied to a specific use case.',
-  'Post-cutover legacy mesh files (mesh_schema.py, mesh_signed_events.py, mesh_hashchain.py) hash-verified against the recorded baseline; the chain extension hook stays surgical.',
+  'Update button no longer throws "admin_session_required" on desktop installs. The initial updateAction now syncs to Tauri detection at React-init time (window.__TAURI__ is injected before mount), so a click before the async runtime probe completes opens the GitHub release page in a browser instead of POSTing to /api/system/update.',
+  'Desktop installer now bundles defusedxml + PySocks (declared in pyproject.toml but missing from the venv shipped with v0.9.79 and the initial v0.9.8 publish). Fixes the bundled-backend launch crash reported in #319 and #296 (managed_backend_exited_early:exit code: 103).',
+  'UAP layer no longer serves 3-year-old NUFORC sightings via the Hugging Face static-mirror fallback (60-day cutoff now applied to the fallback path too).',
+  'GPS jamming detection now counts nac_p == 0 (the actual GPS-lost signal) instead of filtering it out as an old-transponder artifact.',
+  'GPS jamming thresholds lowered (MIN_AIRCRAFT 5 → 3, MIN_RATIO 0.30 → 0.20) so sparser hotspots clear the bar without losing the 1-aircraft noise cushion.',
+  'AIS layer surfaces an outage banner when the AISStream WebSocket upstream is offline, instead of silently showing an empty ocean.',
+  'Flight emissions tooltip now shows cumulative fuel/CO2 since first observation, not just the per-hour rate.',
+  'Per-aircraft observation tracker (15-min reopen gap, 24-hour clamp) survives trail-rendering cache pruning so cumulative counters do not reset mid-flight.',
+  'UAP scheduler moved daily → weekly (Mondays 12:00 UTC) to match the layer’s rolling-window cadence and reduce upstream load.',
 ];
 
 const CONTRIBUTORS = [

frontend/src/components/InfonetTerminal/InfonetShell.tsx

@@ -298,6 +298,33 @@ export default function InfonetShell({
     setCurrentView(view);
   };
 
+  const renderGateDirectory = (variant: 'landing' | 'command' = 'command') => (
+    <div
+      className={
+        variant === 'landing'
+          ? 'w-full max-w-3xl border border-cyan-950/50 bg-black/20 px-4 py-3 text-left shadow-[0_0_18px_rgba(6,182,212,0.06)]'
+          : 'text-gray-400'
+      }
+    >
+      <p className={`${variant === 'landing' ? 'text-[11px]' : ''} text-gray-400 uppercase tracking-[0.18em]`}>
+        AVAILABLE OBFUSCATED GATES:
+      </p>
+      <div className={`grid grid-cols-1 sm:grid-cols-2 xl:grid-cols-3 ${variant === 'landing' ? 'gap-x-8 gap-y-1.5 mt-2' : 'gap-2 mt-2'}`}>
+        {GATES.map(gate => (
+          <button
+            key={gate}
+            type="button"
+            className="group flex min-h-[24px] items-center text-left text-gray-300 hover:text-white transition-colors"
+            onClick={() => handleNavigate('gate', gate)}
+          >
+            <span className="text-gray-500 mr-2 group-hover:text-cyan-400 transition-colors">[{'>'}]</span>
+            <span className="truncate group-hover:drop-shadow-[0_0_5px_rgba(6,182,212,0.8)]">{gate}</span>
+          </button>
+        ))}
+      </div>
+    </div>
+  );
+
   const openGateWhenReady = async (
     gateTarget: string,
     operation: () => Promise<void>,
@@ -471,19 +498,7 @@ export default function InfonetShell({
       setHistory([]);
       return;
     } else if (trimmedCmd === 'gates') {
-      output = (
-        <div className="text-gray-400">
-          <p>AVAILABLE OBFUSCATED GATES:</p>
-          <div className="grid grid-cols-2 md:grid-cols-3 gap-2 mt-2">
-            {GATES.map(gate => (
-              <div key={gate} className="flex items-center cursor-pointer hover:text-gray-300 group" onClick={() => handleNavigate('gate', gate)}>
-                <span className="text-gray-500 mr-2 group-hover:text-cyan-400 transition-colors">[{'>'}]</span>
-                <span className="text-gray-300 group-hover:text-white transition-colors group-hover:drop-shadow-[0_0_5px_rgba(6,182,212,0.8)]">{gate}</span>
-              </div>
-            ))}
-          </div>
-        </div>
-      );
+      output = renderGateDirectory('command');
     } else if (trimmedCmd.startsWith('join ') || trimmedCmd.startsWith('g/')) {
       const target = trimmedCmd.startsWith('g/') ? trimmedCmd.slice(2) : trimmedCmd.split(' ')[1];
       if (GATES.includes(target)) {
@@ -661,6 +676,9 @@ export default function InfonetShell({
                   <p>Type <span className="text-green-400 font-bold">&apos;gates&apos;</span> or <span className="text-green-400 font-bold">g/</span> to view available chatrooms.</p>
                 </div>
                 <NetworkStats />
+                <div className="mt-5 w-full flex justify-center">
+                  {renderGateDirectory('landing')}
+                </div>
               </div>
 
               <HashchainEvents />

frontend/src/components/NewsFeed.tsx

@@ -249,34 +249,70 @@ const VESSEL_TYPE_WIKI: Record<string, string> = {
 
 type FlightTrailPoint = { lat?: number; lng?: number; alt?: number; ts?: number } | number[];
 
+function formatObservedDuration(seconds: number): string {
+    // Compact "1h 14m" / "23m" / "45s" — matches the density of the rest
+    // of the flight tooltip. < 60s is shown as "<1m" so the user knows
+    // we've JUST started observing this hex (cumulative will still be 0).
+    if (!Number.isFinite(seconds) || seconds <= 0) return '<1m';
+    if (seconds < 60) return '<1m';
+    const totalMinutes = Math.floor(seconds / 60);
+    const hours = Math.floor(totalMinutes / 60);
+    const minutes = totalMinutes % 60;
+    if (hours > 0) return `${hours}h ${minutes}m`;
+    return `${minutes}m`;
+}
+
 function EmissionsEstimateBlock({ flight }: { flight: any }) {
     const emissions = flight?.emissions;
-    const context = emissions ? 'Model-based cruise estimate' : null;
+    // Cumulative fuel/CO2 since the backend first saw this hex this
+    // flight session. Prefer these big numbers — the user explicitly
+    // wanted "the actual fuel that has been burned", not the rate.
+    // Rates are still shown below as smaller context.
+    const observedSec = Number(emissions?.observed_seconds ?? 0);
+    const fuelBurned = Number(emissions?.fuel_gallons_burned ?? 0);
+    const co2Emitted = Number(emissions?.co2_kg_emitted ?? 0);
+    const haveCumulative = emissions && observedSec > 0;
 
     return (
         <div className="border-b border-[var(--border-primary)] pb-2">
             <span className="text-[var(--text-muted)] text-[10px] block mb-1.5">EMISSIONS ESTIMATE</span>
             <div className="flex gap-3">
                 <div className="flex-1 bg-[var(--bg-primary)]/50 border border-[var(--border-primary)] px-2 py-1.5">
-                    <div className="text-[11px] text-[var(--text-muted)] tracking-widest">FUEL RATE</div>
-                    <div className="text-xs font-bold text-orange-400">
-                        {emissions ? (
-                            <>{emissions.fuel_gph} <span className="text-[11px] text-[var(--text-muted)] font-normal">GPH</span></>
+                    <div className="text-[11px] text-[var(--text-muted)] tracking-widest">FUEL BURNED</div>
+                    <div className="text-sm font-bold text-orange-400">
+                        {haveCumulative ? (
+                            <>{fuelBurned.toLocaleString(undefined, { maximumFractionDigits: 1 })} <span className="text-[11px] text-[var(--text-muted)] font-normal">gal</span></>
+                        ) : emissions ? (
+                            <span className="text-[var(--text-muted)] font-normal text-xs">—</span>
                         ) : 'UNKNOWN'}
                     </div>
+                    {emissions && (
+                        <div className="text-[10px] text-[var(--text-muted)] mt-0.5">
+                            @ {emissions.fuel_gph} gph
+                        </div>
+                    )}
                 </div>
                 <div className="flex-1 bg-[var(--bg-primary)]/50 border border-[var(--border-primary)] px-2 py-1.5">
-                    <div className="text-[11px] text-[var(--text-muted)] tracking-widest">CO2 RATE</div>
-                    <div className="text-xs font-bold text-red-400">
-                        {emissions ? (
-                            <>{emissions.co2_kg_per_hour.toLocaleString()} <span className="text-[11px] text-[var(--text-muted)] font-normal">KG/HR</span></>
+                    <div className="text-[11px] text-[var(--text-muted)] tracking-widest">CO2 EMITTED</div>
+                    <div className="text-sm font-bold text-red-400">
+                        {haveCumulative ? (
+                            <>{co2Emitted.toLocaleString(undefined, { maximumFractionDigits: 1 })} <span className="text-[11px] text-[var(--text-muted)] font-normal">kg</span></>
+                        ) : emissions ? (
+                            <span className="text-[var(--text-muted)] font-normal text-xs">—</span>
                         ) : 'UNKNOWN'}
                     </div>
+                    {emissions && (
+                        <div className="text-[10px] text-[var(--text-muted)] mt-0.5">
+                            @ {emissions.co2_kg_per_hour.toLocaleString()} kg/hr
+                        </div>
+                    )}
                 </div>
             </div>
-            {context && (
+            {emissions && (
                 <div className="mt-1.5 text-[10px] text-[var(--text-muted)] leading-relaxed">
-                    {context}
+                    {haveCumulative
+                        ? `Observed in flight for ${formatObservedDuration(observedSec)} · model-based cruise estimate`
+                        : 'Just observed · totals will appear on next refresh'}
                 </div>
             )}
         </div>

frontend/src/components/OnboardingModal.tsx

@@ -4,7 +4,7 @@ import React, { useState, useEffect } from 'react';
 import { motion, AnimatePresence } from 'framer-motion';
 import { X, ExternalLink, Key, Shield, Radar, Globe, Satellite, Ship, Radio, Bot, Copy, Check, Network } from 'lucide-react';
 
-const CURRENT_ONBOARDING_VERSION = '0.9.79-agentic-onboarding-1';
+const CURRENT_ONBOARDING_VERSION = '0.9.81-agentic-onboarding-1';
 const STORAGE_KEY = `shadowbroker_onboarding_complete_v${CURRENT_ONBOARDING_VERSION}`;
 const LEGACY_STORAGE_KEY = 'shadowbroker_onboarding_complete';
 

frontend/src/components/StartupWarmupModal.tsx

@@ -4,7 +4,7 @@ import React, { useEffect, useState } from 'react';
 import { motion, AnimatePresence } from 'framer-motion';
 import { Database, Clock, X } from 'lucide-react';
 
-const CURRENT_VERSION = '0.9.79';
+const CURRENT_VERSION = '0.9.81';
 const STORAGE_KEY = `shadowbroker_startup_warmup_notice_v${CURRENT_VERSION}`;
 
 interface StartupWarmupModalProps {

frontend/src/components/TopRightControls.tsx

@@ -91,7 +91,19 @@ export default function TopRightControls({
   const [manualUpdateUrl, setManualUpdateUrl] = useState(DEFAULT_RELEASES_URL);
   const [releasePageUrl, setReleasePageUrl] = useState(DEFAULT_RELEASES_URL);
   const [dockerCommands, setDockerCommands] = useState('');
-  const [updateAction, setUpdateAction] = useState<UpdateActionKind>('auto_apply');
+  // Pre-detection initial value: the right action depends on the runtime.
+  // For desktop installs (Tauri webview), the default should be
+  // ``manual_download`` so that clicking Update before the async runtime
+  // probe completes opens the release page in a browser instead of POSTing
+  // to /api/system/update — which throws ``admin_session_required`` on
+  // fresh sessions and confused v0.9.79/v0.9.8 users with a cryptic error.
+  // ``window.__TAURI__`` is injected synchronously by Tauri before React
+  // mounts, so this check is safe to do at useState init time.
+  const initialUpdateAction: UpdateActionKind =
+    typeof window !== 'undefined' && (window as { __TAURI__?: unknown }).__TAURI__
+      ? 'manual_download'
+      : 'auto_apply';
+  const [updateAction, setUpdateAction] = useState<UpdateActionKind>(initialUpdateAction);
   const [updateDetail, setUpdateDetail] = useState(AUTO_UPDATE_DETAIL);
   const pollRef = useRef<ReturnType<typeof setInterval> | null>(null);
   const timeoutRef = useRef<ReturnType<typeof setTimeout> | null>(null);

frontend/src/hooks/useAisUpstreamHealth.ts

@@ -0,0 +1,85 @@
+/**
+ * useAisUpstreamHealth — polls /api/health and exposes AIS proxy connectivity.
+ *
+ * Background: AISStream's WebSocket server went fully offline 2026-05-23 (TCP
+ * timeouts at stream.aisstream.io). The backend kept reconnecting in a tight
+ * loop and the ships layer silently went empty. Users had no signal that the
+ * problem was upstream, not their config. This hook surfaces the state so a
+ * banner can explain "AIS upstream is offline" instead of letting users
+ * wonder.
+ *
+ * The poll interval is intentionally relaxed (30s) — this is a low-urgency UX
+ * signal, not a real-time data feed. Backend already escalates top_status to
+ * "degraded" when AIS is configured-but-disconnected.
+ */
+import { useEffect, useRef, useState } from 'react';
+import { API_BASE } from '@/lib/api';
+
+export interface AisUpstreamHealth {
+  /** True when we've received a vessel message in the last ~60s. */
+  connected: boolean;
+  /** Seconds since the last vessel message; null when we've never seen one. */
+  lastMsgAgeSeconds: number | null;
+  /**
+   * True when the SPKI-pinned fallback is in effect (issue #258).
+   * Data still flows in this mode — it's a separate, less urgent signal
+   * than ``connected``.
+   */
+  degradedTls: boolean;
+  /** How many times the proxy has been spawned (sustained growth without
+   *  ``connected`` means upstream is dead and we're respawning in a loop). */
+  proxySpawnCount: number;
+  /** Whether the operator has configured an API key. When false, the banner
+   *  shouldn't fire because "AIS is off" is the intended state. The backend
+   *  signals this via the ``connected`` flag being false AND no msg ever
+   *  seen — we approximate it by requiring at least one spawn before
+   *  declaring an outage. */
+  aisEnabled: boolean;
+}
+
+const POLL_INTERVAL_MS = 30_000;
+
+export function useAisUpstreamHealth(): AisUpstreamHealth | null {
+  const [health, setHealth] = useState<AisUpstreamHealth | null>(null);
+  const cancelledRef = useRef(false);
+
+  useEffect(() => {
+    cancelledRef.current = false;
+
+    const fetchHealth = async () => {
+      try {
+        const res = await fetch(`${API_BASE}/api/health`, { cache: 'no-store' });
+        if (!res.ok) return;
+        const body = await res.json();
+        if (cancelledRef.current) return;
+        const proxy = body?.ais_proxy ?? {};
+        // ``proxy_spawn_count > 0`` is the cheapest "AIS is enabled" check:
+        // if the backend never spawned the proxy (no API key, opt-out env)
+        // we shouldn't ever show the outage banner. Once the proxy has
+        // spawned at least once we know the operator wants AIS data.
+        const spawns = Number(proxy.proxy_spawn_count ?? 0);
+        setHealth({
+          connected: Boolean(proxy.connected),
+          lastMsgAgeSeconds:
+            proxy.last_msg_age_seconds == null
+              ? null
+              : Number(proxy.last_msg_age_seconds),
+          degradedTls: Boolean(proxy.degraded_tls),
+          proxySpawnCount: spawns,
+          aisEnabled: spawns > 0,
+        });
+      } catch {
+        // Backend unreachable — separate problem. Banner not relevant.
+      }
+    };
+
+    void fetchHealth();
+    const interval = setInterval(() => void fetchHealth(), POLL_INTERVAL_MS);
+    return () => {
+      cancelledRef.current = true;
+      clearInterval(interval);
+    };
+  }, []);
+
+  return health;
+}

helm/chart/Chart.yaml

@@ -1,8 +1,8 @@
 ---
 apiVersion: v2
 name: shadowbroker
-version: 0.9.79
-appVersion: "0.9.79"
+version: 0.9.81
+appVersion: "0.9.81"
 description: simple shadowbroker installation
 type: application
 

meshnode.bat

@@ -8,6 +8,7 @@ echo ===================================================
 echo.
 echo   Lightweight node — syncs the Infonet chain only.
 echo   No map, no frontend, no data feeds.
+echo   Private hashchain relay: gate messages + offline DM spool.
 echo   Close this window to stop the node.
 echo.
 
@@ -96,15 +97,22 @@ echo [*] Auto-enabling node participation...
 if not exist "data\" mkdir data
 echo {"enabled":true,"updated_at":0} > data\node.json
 
+set MESH_ONLY=true
+set SHADOWBROKER_MESH_NODE_RUNTIME=true
+set MESH_NODE_MODE=participant
+set MESH_INFONET_ALLOW_CLEARNET_SYNC=false
+set MESH_ARTI_ENABLED=true
+set MESH_DM_HASHCHAIN_SPOOL_LIMIT=2
+set MESH_DM_HASHCHAIN_SPOOL_TTL_S=3600
+if "%MESH_BOOTSTRAP_SEED_PEERS%"=="" set MESH_BOOTSTRAP_SEED_PEERS=http://gqpbunqbgtkcqilvclm3xrkt3zowjyl3s62kkktvojgvxzizamvbrqid.onion:8000
+
 echo.
 echo ===================================================
 echo   Mesh node starting on port 8000
 echo   Mode: MESH_ONLY (no data feeds)
-echo   Relay: %MESH_RELAY_PEERS%
+echo   Bootstrap: %MESH_BOOTSTRAP_SEED_PEERS%
 echo   Press Ctrl+C to stop
 echo ===================================================
 echo.
 
-set MESH_ONLY=true
-set MESH_NODE_MODE=participant
 python -m uvicorn main:app --host 0.0.0.0 --port 8000

meshnode.sh

@@ -10,6 +10,7 @@ echo "==================================================="
 echo ""
 echo "  Lightweight node — syncs the Infonet chain only."
 echo "  No map, no frontend, no data feeds."
+echo "  Private hashchain relay: gate messages + offline DM spool."
 echo "  Press Ctrl+C to stop."
 echo ""
 
@@ -51,15 +52,22 @@ echo "[*] Auto-enabling node participation..."
 mkdir -p data
 echo '{"enabled":true,"updated_at":0}' > data/node.json
 
+export MESH_ONLY=true
+export SHADOWBROKER_MESH_NODE_RUNTIME=true
+export MESH_NODE_MODE=participant
+export MESH_INFONET_ALLOW_CLEARNET_SYNC=false
+export MESH_ARTI_ENABLED=true
+export MESH_DM_HASHCHAIN_SPOOL_LIMIT=2
+export MESH_DM_HASHCHAIN_SPOOL_TTL_S=3600
+export MESH_BOOTSTRAP_SEED_PEERS="${MESH_BOOTSTRAP_SEED_PEERS:-http://gqpbunqbgtkcqilvclm3xrkt3zowjyl3s62kkktvojgvxzizamvbrqid.onion:8000}"
+
 echo ""
 echo "==================================================="
 echo "  Mesh node starting on port 8000"
 echo "  Mode: MESH_ONLY (no data feeds)"
-echo "  Relay: ${MESH_RELAY_PEERS:-default}"
+echo "  Bootstrap: ${MESH_BOOTSTRAP_SEED_PEERS}"
 echo "  Press Ctrl+C to stop"
 echo "==================================================="
 echo ""
 
-export MESH_ONLY=true
-export MESH_NODE_MODE=participant
 python3 -m uvicorn main:app --host 0.0.0.0 --port 8000

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "shadowbroker"
-version = "0.9.79"
+version = "0.9.81"
 readme = "README.md"
 requires-python = ">=3.10"
 dependencies = []

start.bat

@@ -237,6 +237,14 @@ echo [*] Backend Node.js dependencies OK.
 echo.
 echo [*] Checking privacy-core shared library...
 set "PRIVACY_CORE_DLL=%ROOT%\privacy-core\target\release\privacy_core.dll"
+:: MSI/EXE installers stage privacy_core.dll directly in backend-runtime/
+:: alongside this script. If somebody runs start.bat from an installed
+:: app directory (no source checkout, no Rust toolchain), they shouldn't
+:: see a spurious "install Rust" warning because the DLL is right next
+:: to them — just at a different path than the source-tree build.
+if not exist "%PRIVACY_CORE_DLL%" if exist "%ROOT%\privacy_core.dll" (
+    set "PRIVACY_CORE_DLL=%ROOT%\privacy_core.dll"
+)
 if not exist "%PRIVACY_CORE_DLL%" (
     where cargo >nul 2>&1
     if errorlevel 1 (

start.sh

@@ -203,6 +203,17 @@ echo ""
 echo "[*] Checking privacy-core shared library..."
 PRIVACY_CORE_SO="$SCRIPT_DIR/privacy-core/target/release/libprivacy_core.so"
 PRIVACY_CORE_DYLIB="$SCRIPT_DIR/privacy-core/target/release/libprivacy_core.dylib"
+# MSI/AppImage/DMG installers stage the platform-specific shared library
+# directly alongside this script (in backend-runtime/). If somebody runs
+# start.sh from an installed app dir without Rust, they shouldn't see a
+# spurious "install Rust" warning — the library is right next to them,
+# just at a different path than the source-tree build.
+if [ ! -f "$PRIVACY_CORE_SO" ] && [ -f "$SCRIPT_DIR/libprivacy_core.so" ]; then
+    PRIVACY_CORE_SO="$SCRIPT_DIR/libprivacy_core.so"
+fi
+if [ ! -f "$PRIVACY_CORE_DYLIB" ] && [ -f "$SCRIPT_DIR/libprivacy_core.dylib" ]; then
+    PRIVACY_CORE_DYLIB="$SCRIPT_DIR/libprivacy_core.dylib"
+fi
 if [ ! -f "$PRIVACY_CORE_SO" ] && [ ! -f "$PRIVACY_CORE_DYLIB" ]; then
     if command -v cargo >/dev/null 2>&1; then
         echo "[*] Building privacy-core release library..."

uv.lock

@@ -74,7 +74,7 @@ wheels = [
 
 [[package]]
 name = "backend"
-version = "0.9.79"
+version = "0.9.81"
 source = { editable = "backend" }
 dependencies = [
     { name = "apscheduler" },
@@ -2231,7 +2231,7 @@ wheels = [
 
 [[package]]
 name = "shadowbroker"
-version = "0.9.79"
+version = "0.9.81"
 source = { virtual = "." }
 
 [package.metadata]