feat(infonet): private gate + DM hashchain spool with hardened propagation (#326)

Private gate messages and offline DMs now ride the Infonet hashchain
as ciphertext-only events, replicated across nodes via private
transports (Tor onion / RNS / loopback) and decrypted only by parties
holding the gate or recipient keys.

Hashchain core (mesh_hashchain.py)
----------------------------------

* New ``append_private_gate_message`` and ``append_private_dm_message``
  append paths with full signature verification, public-key binding,
  revocation check, and replay protection in a dedicated sequence
  domain (so a gate post does not consume the author's public broadcast
  sequence, and a DM cannot replay-block a public message at sequence=1).
* Fork validation and full-chain validation now accept the gate
  signature compatibility variants — older signatures that canonicalize
  with/without epoch or reply_to still verify, so a re-sync from an
  older peer doesn't reject still-valid history.
* DM hashchain spool: capped at 2 active sealed offline DMs per
  recipient mailbox, plus a per-(sender, recipient) cap so one prolific
  sender can't consume both slots. 1-hour TTL on the cap counter.
  Spool intentionally small — it's an offline bootstrap channel,
  not a persistent mailbox.
* Rebuild-state preserves the gate sequence domain across reloads so
  a chain reload doesn't accidentally let an old gate sequence
  replay-collide on next append.

Schema enforcement (mesh_schema.py)
-----------------------------------

* Private gate + DM payloads have closed allowlists of fields.
  Plaintext keys (``message``, ``plaintext``, ``_local_plaintext``,
  ``_local_reply_to``) are explicit rejection-bait — they raise before
  the event ever touches the chain.
* DM ciphertext + nonce must look like base64-ish sealed bytes;
  obvious base64-encoded plaintext shapes are rejected.
* ``transport_lock`` required: DM hashchain spool requires
  ``private_strong``; gate accepts ``private``/``private_strong``/
  ``rns``/``onion``.

Defense-in-depth at the network layer (main.py + mesh_public.py)
----------------------------------------------------------------

* ``_infonet_sync_response_events`` now silently redacts private events
  (gate_message + dm_message) unless the request looks like a loopback /
  onion / RNS / private transport caller. If an operator accidentally
  exposes :8000 to the public internet, an external puller gets
  public events only — never ciphertext.
* ``_sync_from_peer`` raises ``PeerSyncRateLimited`` for 429 (handled
  as 4-tuple return with retry_after_s) and ``PeerSyncHTTPError`` for
  other non-200 statuses (handled by ``_run_public_sync_cycle`` to
  honor server cooldown hints even outside the 429 path).

DM relay hydration (main.py)
-----------------------------

* New ``_hydrate_dm_relay_from_chain``: when accepted dm_message chain
  events arrive on a node, they get deposited into the local DM relay
  store with a deterministic sender_token_hash so re-sync of the same
  event is idempotent. Recipients see the ciphertext as a normal DM
  on their next poll and decrypt with their existing recipient key.

Other surfaces
--------------

* meshnode.bat / meshnode.sh now set ``MESH_INFONET_ALLOW_CLEARNET_SYNC=
  false`` and the participant runtime flags by default so a freshly
  spun-up node defaults to private-only sync.
* InfonetTerminal/InfonetShell.tsx adds a gate directory renderer for
  the new private-gate workflow.
* docker-compose.relay.yml binds the relay backend to 127.0.0.1:8000
  only; Tor's hidden service forwards onion traffic into 127.0.0.1.
  Public clearnet :8000 stays off the network edge.

Tests
-----

* 7 new tests in test_private_gate_hashchain.py + test_private_dm_
  hashchain.py covering: gate fork accepts ciphertext propagation,
  gate fork rejects plaintext, append rejects plaintext before
  normalize, append requires private_strong, append rejects
  non-sealed ciphertext shape, DM spool 2-per-recipient + 1-per-pair
  cap, DM hydration delivers to poll/claim.
* Updated test_mesh_node_bootstrap_runtime.py covers 429 backoff via
  PeerSyncRateLimited 4-tuple AND PeerSyncHTTPError exception.
* Updated test_s14b_public_sync_gate_filter.py + test_s9b_gate_store_
  hydration.py + test_gate_write_cutover.py cover the new private
  redaction on public sync responses.
* test_private_gate_hashchain.py + test_private_dm_hashchain.py:
  10 passed locally.
* Combined mesh-relevant suite (the 5 modified existing tests +
  2 new): 17 passed.

Co-authored-by: Claude Opus 4.7 <noreply@anthropic.com>

backend/main.py

@@ -1,4 +1,4 @@
-import os
+import os
 from dotenv import load_dotenv
 load_dotenv()
 
@@ -8,6 +8,7 @@ import asyncio
 import base64
 import hmac
 import importlib
+import ipaddress
 import secrets
 import hashlib as _hashlib_mod
 from dataclasses import dataclass, field
@@ -20,6 +21,11 @@ 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