relay preparing for public internet

packages/node/meshnet_node/torch_server.py

@@ -169,6 +169,35 @@ class _TorchHTTPServer(http.server.HTTPServer):
         self.queue_depth: int = 0
         self._stats_lock = threading.Lock()
         self._active_requests: dict[str, dict[str, Any]] = {}
+        self._decode_log: dict[str, dict[str, float]] = {}
+
+    def note_decode_step(
+        self, session: str, now: float | None = None,
+    ) -> int | None:
+        """Count one decode forward; return the cumulative step count when a
+        log line is due (first step of a session, then every 5s), else None."""
+        if now is None:
+            now = time.monotonic()
+        with self._stats_lock:
+            rec = self._decode_log.get(session)
+            if rec is None:
+                if len(self._decode_log) >= 64:
+                    stale = [
+                        sid for sid, r in self._decode_log.items()
+                        if now - r["seen"] > 600.0
+                    ]
+                    for sid in stale:
+                        del self._decode_log[sid]
+                    while len(self._decode_log) >= 64:
+                        self._decode_log.pop(next(iter(self._decode_log)))
+                self._decode_log[session] = {"steps": 1.0, "logged": now, "seen": now}
+                return 1
+            rec["steps"] += 1
+            rec["seen"] = now
+            if now - rec["logged"] >= 5.0:
+                rec["logged"] = now
+                return int(rec["steps"])
+            return None
 
     def snapshot_current_requests(self) -> list[dict[str, Any]]:
         """In-flight request snapshots for tracker heartbeats."""
@@ -346,23 +375,34 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler):
         hop_index = int(self.headers.get("X-Meshnet-Hop-Index", "0"))
         if hop_index > 0:
             server.received_activations = True
-        if chunk_index_value == 0:
-            shard_start = getattr(server.backend, "shard_start", "?")
-            shard_end = getattr(server.backend, "shard_end", "?")
-            print(
-                f"  [node] forward hop={hop_index} "
-                f"layers={shard_start}-{shard_end} "
-                f"session={session[:8]}{self._request_log_suffix()}",
-                flush=True,
-            )
-
-        start_layer_header = self.headers.get("X-Meshnet-Start-Layer")
-        start_layer = int(start_layer_header) if start_layer_header else None
-
         # Session KV-cache protocol: prefill establishes per-session state on
         # this node's layer range; decode reuses it. Absent header = legacy
         # stateless call (also the signature fake backends implement).
         cache_mode = self.headers.get("X-Meshnet-Cache")
+        if chunk_index_value == 0:
+            shard_start = getattr(server.backend, "shard_start", "?")
+            shard_end = getattr(server.backend, "shard_end", "?")
+            if cache_mode == "decode":
+                # One decode forward arrives per generated token — log a
+                # periodic per-session summary instead of one line per token.
+                steps = server.note_decode_step(session)
+                if steps is not None:
+                    print(
+                        f"  [node] decoding layers={shard_start}-{shard_end} "
+                        f"session={session[:8]} steps={steps}"
+                        f"{self._request_log_suffix()}",
+                        flush=True,
+                    )
+            else:
+                print(
+                    f"  [node] forward hop={hop_index} "
+                    f"layers={shard_start}-{shard_end} "
+                    f"session={session[:8]}{self._request_log_suffix()}",
+                    flush=True,
+                )
+
+        start_layer_header = self.headers.get("X-Meshnet-Start-Layer")
+        start_layer = int(start_layer_header) if start_layer_header else None
         forward_kwargs: dict[str, object] = {}
         if cache_mode in ("prefill", "decode"):
             past_len_header = self.headers.get("X-Meshnet-Past-Len")
@@ -585,12 +625,14 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler):
 
         session_id = str(uuid.uuid4())
         use_kv = bool(getattr(backend, "supports_kv_cache", False))
+        # EOS detection by id must work on the stateless path too: the tail
+        # returns token_id regardless of caching, and EOS usually decodes to
+        # "" (skip_special_tokens), so the text comparison never fires.
         eos_ids: set[int] = set()
-        if use_kv:
-            try:
-                eos_ids = set(backend.eos_token_ids())
-            except Exception:
-                eos_ids = set()
+        try:
+            eos_ids = set(backend.eos_token_ids())
+        except Exception:
+            eos_ids = set()
 
         stream_emit = None
         if stream: