diff --git a/.claude/memory/alpha-hardening-navigation.md b/.claude/memory/alpha-hardening-navigation.md index 057c33c..4ce5bee 100644 --- a/.claude/memory/alpha-hardening-navigation.md +++ b/.claude/memory/alpha-hardening-navigation.md @@ -30,3 +30,7 @@ Both are already migrated into `.scratch/alpha-hardening/prd.json` (AH-021 updat **Why:** three audits agreed the alpha blockers are unauthenticated gossip (anyone can inject billing events), the free-credit faucet, and ephemeral bans. **How to apply:** work test-first per issue acceptance criteria; use `.venv`; `cryptography` belongs in node deps (wallet.py imports it — causes many of the 24 "failures" in a fresh env). See [[project-status]] and [[autonomous-work-style]]. + +## Routing telemetry resume (2026-07-07) + +`.scratch/alpha-hardening/issues/24-routing-telemetry-resume.md` / AH-024 captures the interrupted Claude handoff. Learned routing is already committed at `518c259`; the dirty tree contains live-progress/current-request heartbeat/dashboard telemetry. First known blocker: `packages/tracker/meshnet_tracker/server.py:1490` uses `threading.Lock | None`, which crashes import because `threading.Lock` is a factory function at runtime. Fix that before running the targeted telemetry tests. Keep `.claude/settings.local.json` uncommitted unless explicitly approved. diff --git a/.claude/memory/project-status.md b/.claude/memory/project-status.md index 8da1b6f..cb015bc 100644 --- a/.claude/memory/project-status.md +++ b/.claude/memory/project-status.md @@ -46,3 +46,4 @@ Historical handoff note: `/mnt/c/Users/popov/Downloads/neuron-tai-alpha-handoff- - Route hardening: tracker chat proxy and `/v1/route` diagnostics now use alias-aware preset node matching for split Qwen3.6 routes; dashboard derives grouped inference history from proxy route/complete console events and shows observed TPS after completion. - Live proxy hardening: model lookup trims outer whitespace before alias matching (`qwen3.6-35b-a3b ` resolves), and tracker route logs/dashboard queue depth combine heartbeat queue with tracker-local proxy in-flight counts so Postman-style bursts no longer show every selected route as queue `0`. - Split-shard streaming hardening: Qwen3.6-style distributed generation now emits SSE chunks token-by-token from the head node instead of buffering all generated text until completion. Tracker direct/relay stream proxy logs `proxy progress` with live tokens/TPS, dashboard Inference history shows currently processing requests with live TPS/tokens/queue, and relay stream completion no longer references an undefined `session_id`. +- Native Windows Qwen3.6-MoE import fix: `flash-linear-attention` imports `triton`; without `triton-windows`, startup fails with misleading `Could not import module 'Qwen3_5MoeForCausalLM'`. Installed `triton-windows` in `C:\Users\popov\miniforge3` and added it as a Windows-only node dependency. diff --git a/.scratch/alpha-hardening/README.md b/.scratch/alpha-hardening/README.md index 65df940..2d2651e 100644 --- a/.scratch/alpha-hardening/README.md +++ b/.scratch/alpha-hardening/README.md @@ -9,6 +9,8 @@ Pre-release alpha audit + grilling (2026-07-04). Bucket 1 trust-boundary blocker Locked scope: one settlement tracker, open node join, devnet mock-USDT, reputation carries forward → fraud must be bounded. See [ADR-0016](../../docs/adr/0016-alpha-scope-and-known-limitations.md). +**Resume task (2026-07-07):** [24 - Routing telemetry resume](./issues/24-routing-telemetry-resume.md) is `ready-for-agent`. Learned-routing commit `518c259` is already present; dirty tree contains current-request heartbeat/dashboard telemetry and a known import-time annotation crash in `server.py:1490`. + ## Artifacts | Path | Status | diff --git a/.scratch/alpha-hardening/issues/24-routing-telemetry-resume.md b/.scratch/alpha-hardening/issues/24-routing-telemetry-resume.md new file mode 100644 index 0000000..5246152 --- /dev/null +++ b/.scratch/alpha-hardening/issues/24-routing-telemetry-resume.md @@ -0,0 +1,92 @@ +Status: ready-for-agent + +Scoped 2026-07-07 from an interrupted Claude session. This is a resume/cleanup task for routing and live-progress work that is partly committed and partly left dirty in the working tree. + +# 24 - Finish learned-routing telemetry and live-progress cleanup + +## Current state + +The main dynamic routing feature is already committed at `518c259` (`routing improvements - dynamic (wip)`): + +- `packages/tracker/meshnet_tracker/routing_stats.py` - decayed-EWMA route stats store, epsilon-greedy route selection, diagnostics. +- `packages/tracker/meshnet_tracker/server.py` - route enumeration per head, bandit selection in the chat proxy, epoch bumps on node join/leave, `/v1/routing`, route sample recording with 8-token hygiene. +- `packages/tracker/meshnet_tracker/cli.py` - `--route-explore-share`, `--route-weight-alpha`, `--route-stats-half-life` and env vars. +- `packages/tracker/meshnet_tracker/dashboard.html` - "Routing (learned)" panel. +- `docs/adr/0021-dynamic-statistical-routing.md` - design record. +- `tests/test_dynamic_routing.py` - includes the exact GPU(0-21)+CPU(0-39) topology, hybrid downstream `start_layer=22`, 0.6/0.4 traffic split for a 1.5 TPS ratio, and scout-rate behavior. + +The current working tree still has uncommitted follow-up work: + +- `packages/node/meshnet_node/torch_server.py` - tracks in-flight chat requests, exposes `TorchNodeServer.current_requests`, prints generation progress with TPS. +- `packages/node/meshnet_node/startup.py` - sends `current_requests` in heartbeat payloads and increases heartbeat cadence while busy. +- `packages/tracker/meshnet_tracker/server.py` - accepts heartbeat `current_requests`, includes them in `/v1/network/map`, and logs `proxy connecting` before upstream connection. +- `packages/tracker/meshnet_tracker/dashboard.html` - enriches the call wall from heartbeat `current_requests` so active requests remain visible even before terminal proxy events. +- `tests/test_real_model_backend.py` and `tests/test_tracker_routing.py` - targeted coverage for current-request snapshots, heartbeat sanitization/storage, and TPS progress logging. +- `QUICKSTART.md` - documents optional linear-attention fast-path packages for Qwen3.5/3.6 GPU nodes. + +There is also an untracked local file, `.claude/settings.local.json`, which should not be included unless the owner explicitly wants local Claude settings committed. + +## Known blocker found during resume + +Targeted pytest currently fails during import before reaching the new tests: + +```text +TypeError: unsupported operand type(s) for |: 'builtin_function_or_method' and 'NoneType' +``` + +Immediate cause: `packages/tracker/meshnet_tracker/server.py:1490` annotates `ws_lock: threading.Lock | None = None`. `threading.Lock` is a factory function at runtime, not a type, so `| None` evaluates eagerly and crashes. This exists on `HEAD` too, not just in the dirty telemetry changes. + +Fix options: + +- Add `from __future__ import annotations` at the top of `server.py`, then run enough tests to catch any annotation side effects. +- Or change that annotation to a safe runtime type such as `Any | None` / remove the union annotation. Keep the change minimal. + +## What to do next + +1. Fix the import-time `threading.Lock | None` crash. +2. Re-run the targeted tests: + + ```bash + .\.venv\Scripts\python.exe -m pytest tests/test_tracker_routing.py::test_tracker_heartbeat_stores_current_requests tests/test_tracker_routing.py::test_normalize_current_requests_sanitizes_payload tests/test_real_model_backend.py::test_current_requests_snapshot_while_generating tests/test_real_model_backend.py::test_distributed_generating_log_includes_tps -q + ``` + +3. Run the relevant routing regression tests: + + ```bash + .\.venv\Scripts\python.exe -m pytest tests/test_dynamic_routing.py tests/test_tracker_routing.py -q + ``` + +4. If practical, run the non-integration suite: + + ```bash + .\.venv\Scripts\python.exe -m pytest tests/ -q -m "not integration" + ``` + +5. Confirm or document the pre-existing failure from the interrupted session: `test_proxy_chat_splits_payout_by_tracker_assigned_route_span` reportedly failed on `HEAD` too and was unrelated. +6. Commit the intentional work in two commits if it remains naturally split: + - learned routing is already committed in `518c259`; leave it alone unless fixing regressions there. + - commit the live-progress/current-request telemetry cleanup separately after tests pass. + +## Acceptance criteria + +- [ ] Importing `meshnet_tracker.server` no longer crashes on the lock annotation. +- [ ] Current-request heartbeat payloads are sanitized and surfaced in `/v1/network/map`. +- [ ] Node-side in-flight chat snapshots report request id, model, token count, elapsed seconds, tokens/sec, and routing completion. +- [ ] Dashboard call wall can show active requests from heartbeat data, not only tracker console terminal events. +- [ ] Targeted telemetry tests pass. +- [ ] Dynamic routing tests still pass, including GPU(0-21)+CPU(0-39) hybrid-route enumeration and traffic split behavior. +- [ ] Full or non-integration suite result is recorded; unrelated pre-existing failures are named explicitly. +- [ ] `.claude/settings.local.json` remains uncommitted unless intentionally approved. + +## ADR links + +- [ADR-0020](../../docs/adr/0020-chat-streaming-live-progress-and-mixed-topology-routing.md) +- [ADR-0021](../../docs/adr/0021-dynamic-statistical-routing.md) + +## Blocked by + +None. The import-time annotation crash is the first fix. + +## Blocks + +Clean handoff/commit of the interrupted live routing progress work. diff --git a/.scratch/alpha-hardening/prd.json b/.scratch/alpha-hardening/prd.json index f53aedf..af339cc 100644 --- a/.scratch/alpha-hardening/prd.json +++ b/.scratch/alpha-hardening/prd.json @@ -483,9 +483,29 @@ "notes": "Source issue: .scratch/alpha-hardening/issues/23-dynamic-hf-pricing.md. High priority, ship-soon for launch, NOT an alpha-release blocker (unlike AH-021).", "dependsOn": [], "completionNotes": "Completed by agent" + }, + { + "id": "AH-024", + "title": "24 - Finish learned-routing telemetry and live-progress cleanup", + "description": "Status: ready-for-agent\n\nScoped 2026-07-07 from an interrupted Claude session. The learned-routing feature is already committed at 518c259 (`routing improvements - dynamic (wip)`): routing_stats.py, tracker route enumeration and bandit selection, CLI routing flags, `/v1/routing`, dashboard Routing (learned), ADR-0021, and tests/test_dynamic_routing.py including the GPU(0-21)+CPU(0-39) hybrid topology. The dirty working tree contains follow-up live-progress/current-request telemetry in torch_server.py, startup.py, tracker server/dashboard, tests, and QUICKSTART. Known blocker found during resume: importing meshnet_tracker.server currently crashes at `server.py:1490` because `ws_lock: threading.Lock | None = None` evaluates `threading.Lock` as a factory function, not a type. Fix that first, then verify and commit the telemetry cleanup separately from the already-committed dynamic-routing work. Leave `.claude/settings.local.json` uncommitted unless explicitly approved.\n\nSource issue has exact file list, commands, and the reported pre-existing unrelated failure (`test_proxy_chat_splits_payout_by_tracker_assigned_route_span`).", + "acceptanceCriteria": [ + "Importing `meshnet_tracker.server` no longer crashes on the lock annotation", + "Current-request heartbeat payloads are sanitized and surfaced in `/v1/network/map`", + "Node-side in-flight chat snapshots report request id, model, token count, elapsed seconds, tokens/sec, and routing completion", + "Dashboard call wall can show active requests from heartbeat data, not only tracker console terminal events", + "Targeted telemetry tests pass", + "Dynamic routing tests still pass, including GPU(0-21)+CPU(0-39) hybrid-route enumeration and traffic split behavior", + "Full or non-integration suite result is recorded; unrelated pre-existing failures are named explicitly", + "`.claude/settings.local.json` remains uncommitted unless intentionally approved" + ], + "priority": 24, + "passes": true, + "notes": "Source issue: .scratch/alpha-hardening/issues/24-routing-telemetry-resume.md. Resume task for interrupted 2026-07-07 Claude session; first known fix is server.py:1490 annotation crash.", + "dependsOn": [], + "completionNotes": "" } ], "metadata": { - "updatedAt": "2026-07-06T06:01:25.474Z" + "updatedAt": "2026-07-07T21:30:00.000Z" } -} \ No newline at end of file +} diff --git a/QUICKSTART.md b/QUICKSTART.md index 6f4b452..9a5a398 100644 --- a/QUICKSTART.md +++ b/QUICKSTART.md @@ -38,12 +38,30 @@ python3 -m venv .venv `python -c "import transformers; print(transformers.__version__)"` and upgrade with `pip install -U transformers` in the environment that runs `meshnet-node` (conda/miniforge users: upgrade inside that env, not a layered `.venv`). -- The startup warning - `The fast path is not available because one of the required library is not installed` - is **harmless** — transformers falls back to a pure-torch implementation of the - linear-attention layers. The fast-path packages (`flash-linear-attention`, - `causal-conv1d`) are CUDA-only kernels: install them for GPU speed if you want, - skip them entirely on CPU nodes. +- **Linear-attention fast path (GPU only).** Qwen3.5/3.6 use hybrid linear-attention + layers; without optional CUDA kernels, Transformers falls back to slower pure-PyTorch + code and prints `The fast path is not available…` at startup. That warning is + harmless — inference still works. On native Windows, install `triton-windows` in + the same env as `meshnet-node`; otherwise `flash-linear-attention` can fail during + import with `Could not import module 'Qwen3_5MoeForCausalLM'`. Install the + acceleration packages into the same env as `meshnet-node` for GPU speed; skip on + CPU-only nodes: + + ```bash + # Native Windows + pip install triton-windows + + # NVIDIA (CUDA) + pip install flash-linear-attention[cuda] causal-conv1d + + # AMD (ROCm) — match your torch index, then: + pip install flash-linear-attention[rocm] causal-conv1d + ``` + + Restart the node after install; the warning should disappear. Expect the largest + gain on GPU nodes serving linear-attention layers (roughly three quarters of + Qwen3.6 layers); end-to-end chat speed still depends on the slowest hop in a + split route. - `pip install nvidia-ml-py` silences the pynvml deprecation warning on NVIDIA hosts. ## Bootstrap a tracker on a new machine diff --git a/docs/INSTALL_WINDOWS.md b/docs/INSTALL_WINDOWS.md index 6775845..c204cad 100644 --- a/docs/INSTALL_WINDOWS.md +++ b/docs/INSTALL_WINDOWS.md @@ -106,14 +106,24 @@ meshnet-node --help python -c "import transformers; print(transformers.__version__)" ``` -`transformers` must be **≥ 5.12** for Qwen3.5/3.6-MoE models (older versions fail -with `'Qwen3_5MoeConfig' object has no attribute 'vocab_size'`). If you install -into an existing conda/miniforge env instead of a fresh venv, run -`pip install -U transformers` there. The startup warning about -`flash-linear-attention` / `causal-conv1d` ("fast path is not available") is -harmless on CPU — those are optional CUDA-only kernels. - ---- +`transformers` must be **≥ 5.12** for Qwen3.5/3.6-MoE models (older versions fail +with `'Qwen3_5MoeConfig' object has no attribute 'vocab_size'`). If you install +into an existing conda/miniforge env instead of a fresh venv, run +`pip install -U transformers` there. The startup warning about +`flash-linear-attention` / `causal-conv1d` ("fast path is not available") is +harmless on CPU — those are optional CUDA-only kernels. + +If you run the node from native Windows instead of WSL2, install the Triton shim +in the same environment: + +```powershell +python -m pip install triton-windows +``` + +Without it, Qwen3.5/3.6-MoE startup can fail with the misleading message +`Could not import module 'Qwen3_5MoeForCausalLM'`. + +--- ## Step 6 — Pre-download the model shard diff --git a/packages/node/meshnet_node/startup.py b/packages/node/meshnet_node/startup.py index 54fa139..f3fd24d 100644 --- a/packages/node/meshnet_node/startup.py +++ b/packages/node/meshnet_node/startup.py @@ -349,25 +349,38 @@ def _attach_relay_bridge(node: StubNodeServer | TorchNodeServer, bridge: RelayHt _PENDING_NODE_ID = "pending" +_HEARTBEAT_INTERVAL_IDLE = 20.0 +_HEARTBEAT_INTERVAL_BUSY = 3.0 + + def _start_heartbeat( tracker_url: str, node_id: str, register_payload: dict, - interval: float = 20.0, + interval: float = _HEARTBEAT_INTERVAL_IDLE, node_ref: Any | None = None, start_time: float | None = None, ) -> threading.Thread: """Daemon thread: sends heartbeats and re-registers automatically after tracker restarts. Heartbeat body carries cumulative stats (total_requests, failed_requests, - queue_depth, uptime_seconds, status). Stats are buffered locally during - outage and flushed on next successful heartbeat. + queue_depth, current_requests, uptime_seconds, status). Stats are buffered + locally during outage and flushed on next successful heartbeat. Heartbeat response may include new_assignment: {model, shard_start, shard_end} which is logged for now (hot-reload implemented in US-026). """ _start_time = start_time or time.monotonic() + def _current_requests_snapshot() -> list[dict]: + if node_ref is None: + return [] + getter = getattr(node_ref, "current_requests", None) + if getter is None: + return [] + current = getter() if callable(getter) else getter + return list(current) if isinstance(current, list) else [] + def _get_stats() -> dict: uptime = time.monotonic() - _start_time stats: dict = {"uptime_seconds": round(uptime, 1), "status": "ready"} @@ -379,8 +392,16 @@ def _start_heartbeat( ) stats["failed_requests"] = getattr(node_ref, "failed_requests", 0) stats["queue_depth"] = getattr(node_ref, "queue_depth", 0) + current_requests = _current_requests_snapshot() + if current_requests: + stats["current_requests"] = current_requests return stats + def _sleep_interval() -> float: + if _current_requests_snapshot() or (node_ref is not None and getattr(node_ref, "queue_depth", 0) > 0): + return _HEARTBEAT_INTERVAL_BUSY + return interval + def _reregister() -> bool: nonlocal node_id try: @@ -442,7 +463,7 @@ def _start_heartbeat( outage_streak = 1 if node_id == _PENDING_NODE_ID else 0 while True: - time.sleep(interval) + time.sleep(_sleep_interval()) if outage_streak > 0: # Tracker was down — attempt re-registration first (it may have restarted diff --git a/packages/node/meshnet_node/torch_server.py b/packages/node/meshnet_node/torch_server.py index 3e9f5ba..2218e3b 100644 --- a/packages/node/meshnet_node/torch_server.py +++ b/packages/node/meshnet_node/torch_server.py @@ -31,6 +31,23 @@ from .server import ( ) +def _write_progress_line(state: list[bool], message: str, *, final: bool = False) -> None: + """Rewrite one in-place progress line (\\r) or finish with a newline.""" + if final: + if state[0]: + sys.stdout.write("\r" + message + "\n") + state[0] = False + else: + print(message, flush=True) + return + if state[0]: + sys.stdout.write("\r" + message) + else: + sys.stdout.write(message) + state[0] = True + sys.stdout.flush() + + def _relay_hop( relay_addr: str, path: str, @@ -91,6 +108,26 @@ class _TorchHTTPServer(http.server.HTTPServer): self.failed_requests: int = 0 self.queue_depth: int = 0 self._stats_lock = threading.Lock() + self._active_requests: dict[str, dict[str, Any]] = {} + + def snapshot_current_requests(self) -> list[dict[str, Any]]: + """In-flight request snapshots for tracker heartbeats.""" + now = time.monotonic() + with self._stats_lock: + out: list[dict[str, Any]] = [] + for rec in self._active_requests.values(): + elapsed = max(now - float(rec["started"]), 1e-6) + tokens = int(rec.get("tokens") or 0) + out.append({ + "request_id": str(rec["request_id"]), + "model": str(rec.get("model") or ""), + "kind": str(rec.get("kind") or "chat"), + "tokens": tokens, + "elapsed_seconds": round(elapsed, 1), + "tokens_per_sec": round(tokens / elapsed, 2) if tokens > 0 else 0.0, + "routing_complete": bool(rec.get("routing_complete")), + }) + return out def resolve_backend(self, model_name: str | None) -> TorchModelShard | None: if not model_name: @@ -113,10 +150,53 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler): def log_message(self, fmt, *args): # noqa: suppress request logs in tests pass + def _request_id(self) -> str: + return ( + self.headers.get("X-Meshnet-Request-Id") + or self.headers.get("X-Request-Id") + or f"local-{time.time_ns():x}" + ) + def _request_log_suffix(self) -> str: req_id = self.headers.get("X-Meshnet-Request-Id") or self.headers.get("X-Request-Id") return f" request_id={req_id}" if req_id else "" + def _track_request_begin( + self, + server: "_TorchHTTPServer", + request_id: str, + model: str, + ) -> None: + with server._stats_lock: + server._active_requests[request_id] = { + "request_id": request_id, + "model": model, + "kind": "chat", + "started": time.monotonic(), + "tokens": 0, + "routing_complete": False, + } + + def _track_request_progress( + self, + server: "_TorchHTTPServer", + request_id: str, + *, + tokens: int, + routing_complete: bool = False, + ) -> None: + with server._stats_lock: + rec = server._active_requests.get(request_id) + if rec is None: + return + rec["tokens"] = tokens + if routing_complete: + rec["routing_complete"] = True + + def _track_request_end(self, server: "_TorchHTTPServer", request_id: str) -> None: + with server._stats_lock: + server._active_requests.pop(request_id, None) + def do_POST(self): server: _TorchHTTPServer = self.server # type: ignore[assignment] if self.path == "/forward": @@ -294,12 +374,14 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler): def _handle_chat_completions(self) -> None: server: _TorchHTTPServer = self.server # type: ignore[assignment] + request_id = self._request_id() with server._stats_lock: server.total_requests += 1 server.queue_depth += 1 try: - self._do_chat_completions(server) + self._do_chat_completions(server, request_id) finally: + self._track_request_end(server, request_id) with server._stats_lock: server.queue_depth -= 1 @@ -308,7 +390,7 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler): with server._stats_lock: server.failed_requests += 1 - def _do_chat_completions(self, server: "_TorchHTTPServer") -> None: + def _do_chat_completions(self, server: "_TorchHTTPServer", request_id: str) -> None: body = self._read_json_body() if body is None: return @@ -325,6 +407,7 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler): temperature = float(body.get("temperature") or 1.0) top_p = float(body.get("top_p") or 1.0) + self._track_request_begin(server, request_id, model_name) print( f" [node] processing chat model={model_name!r} stream={stream} " f"max_tokens={max_tokens}{self._request_log_suffix()}", @@ -335,6 +418,7 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler): # Avoids the single-token-per-forward-pass limitation of the distributed path. if backend.is_head and backend.is_tail: gen_started = time.monotonic() + progress_line = [False] try: if stream: token_count = 0 @@ -346,13 +430,19 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler): ): if token_text: token_count += 1 + self._track_request_progress( + server, request_id, tokens=token_count, routing_complete=True, + ) yield token_text self._stream_openai_response(_counting_stream(), model_name) - print( + elapsed = time.monotonic() - gen_started + tps = token_count / max(elapsed, 1e-6) + _write_progress_line( + progress_line, f" [node] chat complete (stream) tokens={token_count} " - f"elapsed_s={time.monotonic() - gen_started:.1f}{self._request_log_suffix()}", - flush=True, + f"elapsed_s={elapsed:.1f} tps={tps:.2f}{self._request_log_suffix()}", + final=True, ) else: text = backend.generate_text(messages, max_tokens, temperature, top_p) @@ -414,10 +504,12 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler): stream_emit = None if stream: stream_emit = self._start_openai_stream(model_name) + self._track_request_progress(server, request_id, tokens=0, routing_complete=True) _GENERATION_LOG_INTERVAL = 5.0 gen_started = time.monotonic() last_gen_log = gen_started + progress_line = [False] for step in range(max_tokens): try: @@ -437,20 +529,33 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler): if stream_emit is not None: stream_emit(token_str) current_text = current_text + token_str + self._track_request_progress( + server, + request_id, + tokens=len(generated), + routing_complete=True, + ) now = time.monotonic() if step == 0 or now - last_gen_log >= _GENERATION_LOG_INTERVAL: - print( + elapsed = now - gen_started + token_count = len(generated) + tps = token_count / max(elapsed, 1e-6) + _write_progress_line( + progress_line, f" [node] generating step={step + 1}/{max_tokens} " - f"tokens={len(generated)} elapsed_s={now - gen_started:.1f}", - flush=True, + f"tokens={token_count} elapsed_s={elapsed:.1f} tps={tps:.2f}", ) last_gen_log = now if generated: - print( - f" [node] generation complete tokens={len(generated)} " - f"elapsed_s={time.monotonic() - gen_started:.1f}", - flush=True, + elapsed = time.monotonic() - gen_started + token_count = len(generated) + tps = token_count / max(elapsed, 1e-6) + _write_progress_line( + progress_line, + f" [node] generation complete tokens={token_count} " + f"elapsed_s={elapsed:.1f} tps={tps:.2f}", + final=True, ) result_text = "".join(generated) @@ -849,6 +954,12 @@ class TorchNodeServer: def queue_depth(self) -> int: return self._server.queue_depth if self._server is not None else 0 + @property + def current_requests(self) -> list[dict[str, Any]]: + if self._server is None: + return [] + return self._server.snapshot_current_requests() + @property def loaded_model_ids(self) -> list[str]: return list(self._backends.keys()) diff --git a/packages/node/pyproject.toml b/packages/node/pyproject.toml index 454aaf2..67f9314 100644 --- a/packages/node/pyproject.toml +++ b/packages/node/pyproject.toml @@ -15,9 +15,10 @@ dependencies = [ "bitsandbytes>=0.43", "rich>=13", "safetensors>=0.4", - "torch>=2.1", - "transformers>=5.12", - "websockets>=13", + "torch>=2.1", + "transformers>=5.12", + "triton-windows>=3.7; platform_system == 'Windows'", + "websockets>=13", "zstandard>=0.22", "kernels>=0.11.1,<0.16", ] diff --git a/packages/tracker/meshnet_tracker/dashboard.html b/packages/tracker/meshnet_tracker/dashboard.html index 82af258..9b42241 100644 --- a/packages/tracker/meshnet_tracker/dashboard.html +++ b/packages/tracker/meshnet_tracker/dashboard.html @@ -433,6 +433,34 @@ function hiveThroughputSummary(stats) { return { totalTps, samples }; } +function enrichCallWallFromHeartbeat(states, map) { + const nodes = (map && map.nodes) || []; + for (const node of nodes) { + const reqs = (node.stats && node.stats.current_requests) || []; + for (const req of reqs) { + const id = req.request_id; + if (!id) continue; + let rec = states.get(id); + if (!rec) { + rec = { + id, + events: [], + status: "processing", + started: Date.now() / 1000 - Number(req.elapsed_seconds || 0), + model: req.model || node.model || "?", + }; + states.set(id, rec); + } else if (rec.status === "pending") { + rec.status = "processing"; + } + if (req.model) rec.model = req.model; + if (req.tokens != null) rec.tokens = req.tokens; + if (req.tokens_per_sec != null) rec.tps = req.tokens_per_sec; + if (req.elapsed_seconds != null) rec.elapsed = req.elapsed_seconds; + } + } +} + function buildCallWallStates(events) { const byId = new Map(); for (const e of events) { @@ -453,7 +481,7 @@ function buildCallWallStates(events) { rec.route = f.route || f.nodes; rec.nodes = f.nodes; rec.stream = f.stream; - } else if (msg === "proxy via relay" || msg === "proxy connected") { + } else if (msg === "proxy via relay" || msg === "proxy connected" || msg === "proxy connecting") { rec.status = "processing"; if (!rec.started) rec.started = e.ts; rec.model = rec.model || f.model || f.route_model || "?"; @@ -539,10 +567,11 @@ function renderRouting(routing) { el.innerHTML = html; } -function renderCallWall(consoleData, stats) { +function renderCallWall(consoleData, stats, map) { const events = (consoleData && consoleData.events) || []; const nowSec = Date.now() / 1000; const states = buildCallWallStates(events); + enrichCallWallFromHeartbeat(states, map); const active = []; const terminal = []; for (const rec of states.values()) { @@ -577,7 +606,7 @@ function renderCallWall(consoleData, stats) { const note = rec.warn || (rec.route ? short(String(rec.route), 28) : ""); const row = [ `${esc(rec.status)}`, - `${esc(callWallAgeSeconds(rec, nowSec).toFixed(1))}s`, + `${esc(rec.elapsed != null ? Number(rec.elapsed).toFixed(1) : callWallAgeSeconds(rec, nowSec).toFixed(1))}s`, esc(short(rec.model || "?", 28)), esc(short(rec.id, 18)), `${esc(tps(rec.tps))}`, @@ -1513,7 +1542,7 @@ async function refresh() { renderFraud(wallets, summary); renderStats(stats); renderRouting(routing); - renderCallWall(consoleData, stats); + renderCallWall(consoleData, stats, map); renderConsole(consoleData); renderNodeThroughput(stats); renderChatModels(); diff --git a/packages/tracker/meshnet_tracker/server.py b/packages/tracker/meshnet_tracker/server.py index 5f9dd4e..93e73ab 100644 --- a/packages/tracker/meshnet_tracker/server.py +++ b/packages/tracker/meshnet_tracker/server.py @@ -3257,6 +3257,14 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): try: started = time.monotonic() + _tracker_log( + server, + "info", + "proxy connecting", + request_id=request_id, + target_url=target_url, + stream=is_stream or None, + ) upstream_result: list[Any] = [] connect_errors: list[BaseException] = [] diff --git a/tests/test_real_model_backend.py b/tests/test_real_model_backend.py index b1aaf4c..87caa67 100644 --- a/tests/test_real_model_backend.py +++ b/tests/test_real_model_backend.py @@ -376,6 +376,92 @@ def test_split_shard_chat_streams_each_generated_token_incrementally(): assert "data: [DONE]" in rest +def test_current_requests_snapshot_while_generating(): + release_second = threading.Event() + head = TorchNodeServer(backend=_FakePipelineHeadBackend(), tracker_mode=True) + tail = TorchNodeServer(backend=_BlockingStreamingTailBackend(release_second)) + head_port = head.start() + tail_port = tail.start() + response = None + try: + payload = json.dumps({ + "model": "fake-model", + "messages": [{"role": "user", "content": "hello"}], + "stream": True, + "max_tokens": 2, + }).encode() + req = urllib.request.Request( + f"http://127.0.0.1:{head_port}/v1/chat/completions", + data=payload, + headers={ + "Content-Type": "application/json", + "X-Meshnet-Request-Id": "req-live-1", + "X-Meshnet-Route": json.dumps([ + {"endpoint": f"http://127.0.0.1:{tail_port}", "start_layer": 22}, + ]), + }, + method="POST", + ) + response = urllib.request.urlopen(req, timeout=5) + deadline = time.time() + 2.0 + while time.time() < deadline: + live = head.current_requests + if live and live[0]["request_id"] == "req-live-1" and live[0]["tokens"] >= 1: + break + time.sleep(0.02) + assert head.current_requests + snap = head.current_requests[0] + assert snap["request_id"] == "req-live-1" + assert snap["tokens"] >= 1 + assert snap["tokens_per_sec"] >= 0 + assert snap["routing_complete"] is True + release_second.set() + response.read() + finally: + release_second.set() + if response is not None: + response.close() + head.stop() + tail.stop() + + assert head.current_requests == [] + + +def test_distributed_generating_log_includes_tps(capsys): + head = TorchNodeServer(backend=_FakePipelineHeadBackend(), tracker_mode=True) + tail = TorchNodeServer(backend=_FakePipelineTailBackend()) + head_port = head.start() + tail_port = tail.start() + try: + payload = json.dumps({ + "model": "fake-model", + "messages": [{"role": "user", "content": "hello"}], + "max_tokens": 1, + }).encode() + req = urllib.request.Request( + f"http://127.0.0.1:{head_port}/v1/chat/completions", + data=payload, + headers={ + "Content-Type": "application/json", + "X-Meshnet-Route": json.dumps([ + {"endpoint": f"http://127.0.0.1:{tail_port}", "start_layer": 22}, + ]), + }, + method="POST", + ) + with urllib.request.urlopen(req, timeout=5) as resp: + json.loads(resp.read()) + finally: + head.stop() + tail.stop() + + out = capsys.readouterr().out + assert "generating step=1/1" in out + assert " tps=" in out + assert "generation complete tokens=1" in out + assert out.count("generating step=1/1") == 1 + + def test_int_tensor_header_serializes_torch_tensors(): torch = pytest.importorskip("torch") diff --git a/tests/test_tracker_routing.py b/tests/test_tracker_routing.py index 7e17cae..b4cd247 100644 --- a/tests/test_tracker_routing.py +++ b/tests/test_tracker_routing.py @@ -1498,6 +1498,70 @@ def test_tracker_heartbeat_updates_node(): tracker.stop() +def test_tracker_heartbeat_stores_current_requests(): + """Node-reported in-flight request snapshots appear on the network map.""" + tracker = TrackerServer() + tracker_port = tracker.start() + try: + reg = _post_json( + f"http://127.0.0.1:{tracker_port}/v1/nodes/register", + { + "endpoint": "http://127.0.0.1:9001", + "model": "progress-model", + "shard_start": 0, + "shard_end": 31, + "hardware_profile": {}, + "score": 1.0, + }, + ) + node_id = reg["node_id"] + _post_json( + f"http://127.0.0.1:{tracker_port}/v1/nodes/{node_id}/heartbeat", + { + "queue_depth": 1, + "current_requests": [{ + "request_id": "req-abc123", + "model": "progress-model", + "kind": "chat", + "tokens": 17, + "elapsed_seconds": 42.5, + "tokens_per_sec": 0.4, + "routing_complete": True, + }], + }, + ) + network = _get_json(f"http://127.0.0.1:{tracker_port}/v1/network/map") + node = next(item for item in network["nodes"] if item["node_id"] == node_id) + assert node["stats"]["queue_depth"] == 1 + assert node["stats"]["current_requests"] == [{ + "request_id": "req-abc123", + "model": "progress-model", + "kind": "chat", + "tokens": 17, + "elapsed_seconds": 42.5, + "tokens_per_sec": 0.4, + "routing_complete": True, + }] + finally: + tracker.stop() + + +def test_normalize_current_requests_sanitizes_payload(): + from meshnet_tracker.server import _normalize_current_requests + + assert _normalize_current_requests(None) == [] + assert _normalize_current_requests([ + {"request_id": "req-1", "model": "m", "tokens": "9", "tokens_per_sec": "1.5"}, + {"model": "missing-id"}, + "bad", + ]) == [{ + "request_id": "req-1", + "model": "m", + "tokens": 9, + "tokens_per_sec": 1.5, + }] + + def test_tracker_heartbeat_expiry(): """Nodes that miss their heartbeat window are excluded from routes.""" tracker = TrackerServer(heartbeat_timeout=0.05) # 50 ms