20 Commits

Author SHA1 Message Date
Dobromir Popov
f0dc3bd93f try to fix streaming responses 2026-07-07 22:19:22 +03:00
Dobromir Popov
a0b37ad1b9 store sessions in the DB 2026-07-07 22:13:12 +03:00
Dobromir Popov
dae0719a32 logging, routing 2026-07-07 22:00:54 +03:00
Dobromir Popov
481ce6c6f5 Merge branch 'master' of https://git.d-popov.com/popov/neuron-tai 2026-07-07 19:20:37 +02:00
Dobromir Popov
7ba87051f5 Document transformers>=5.12 requirement and Qwen3.5/3.6-MoE fast-path notes
Bump the node package's transformers floor to 5.12 (older versions lack
composite Qwen3_5MoeConfig handling and fail with missing vocab_size), and
explain in QUICKSTART/INSTALL_WINDOWS that the flash-linear-attention /
causal-conv1d fast-path warning is a harmless CPU fallback.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-07 19:18:51 +02:00
Dobromir Popov
ac0ca20b56 Merge branch 'master' of https://git.d-popov.com/popov/neuron-tai 2026-07-07 20:16:40 +03:00
Dobromir Popov
38355eba25 innore 2026-07-07 20:16:39 +03:00
Dobromir Popov
471893c9d5 Skip multimodal/MTP checkpoint tensors absent from the text-only causal LM
Qwen3.5/3.6-MoE checkpoints ship vision (model.visual.*) and multi-token-
prediction (mtp.*) weights; the partial shard loader assigned them into the
text-only Qwen3_5MoeForCausalLM and crashed with AttributeError 'mtp'.
Filter selected tensors against the built model's state_dict keys, matching
transformers' _keys_to_ignore_on_load_unexpected behavior.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-07 19:16:19 +02:00
Dobromir Popov
a0dcbfbfd0 Merge branch 'master' of https://git.d-popov.com/popov/neuron-tai 2026-07-07 18:56:10 +02:00
Dobromir Popov
0d8162dcd3 fix xhat 2026-07-07 18:56:08 +02:00
Dobromir Popov
3fc8228590 ignore 2026-07-07 19:46:32 +03:00
Dobromir Popov
6374082b1b Merge branch 'master' of https://git.d-popov.com/popov/neuron-tai 2026-07-07 19:42:40 +03:00
Dobromir Popov
16614855bc new chat layout 2026-07-07 18:42:05 +02:00
Dobromir Popov
cdd2699e63 try fix model loading quen3.6-35b 2026-07-07 18:36:29 +02:00
Dobromir Popov
912ee4f1fd db 2026-07-07 19:31:29 +03:00
Dobromir Popov
f1eea5b6d4 Redesign tracker chat UI with session sidebar and browser-local history. 2026-07-07 18:25:32 +02:00
Dobromir Popov
456c43ea1d set max tokens to 5k 2026-07-07 18:21:13 +02:00
Dobromir Popov
aba5fb12fa Log node request processing so operators can see live activity in the console. 2026-07-07 18:12:57 +02:00
Dobromir Popov
1eb1e0baa2 Merge branch cursor/fix-meshnet-node-param-parsing into master.
Combine shard label formatting with model/shard flag parsing and tracker registration retry.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-07-07 18:02:01 +02:00
Dobromir Popov
b1f08c45cd misc 2026-07-07 18:49:32 +03:00
25 changed files with 4600 additions and 2643 deletions

Binary file not shown.

10
.gitignore vendored
View File

@@ -18,5 +18,11 @@ dist/
!.env.example !.env.example
!.env.testnet !.env.testnet
.rocm-local/* .rocm-local/*
billing.sqlite .pytest-tmp/*
.pytest-tmp/*
# Local tracker/node sqlite databases (never commit runtime state)
*.sqlite
*.sqlite3
logs/tracker/error.log
logs/tracker/info.log
logs/tracker/warning.log

File diff suppressed because it is too large Load Diff

Binary file not shown.

Binary file not shown.

View File

@@ -1,212 +1,220 @@
# Installing meshnet-node on Windows 11 with WSL2 # Installing meshnet-node on Windows 11 with WSL2
This guide covers setting up a meshnet-node on a Windows 11 machine using WSL2 with CUDA passthrough so it can join an existing inference network over LAN. This guide covers setting up a meshnet-node on a Windows 11 machine using WSL2 with CUDA passthrough so it can join an existing inference network over LAN.
## Prerequisites ## Prerequisites
- Windows 11 with WSL2 support (most systems with Windows 10 version 2004+ qualify) - Windows 11 with WSL2 support (most systems with Windows 10 version 2004+ qualify)
- NVIDIA GPU with CUDA support (driver ≥ 525.x recommended for WSL2 CUDA) - NVIDIA GPU with CUDA support (driver ≥ 525.x recommended for WSL2 CUDA)
- At least 8 GB RAM + enough VRAM for the model shard you intend to serve - At least 8 GB RAM + enough VRAM for the model shard you intend to serve
- The Linux machine (other node) is reachable on your LAN - The Linux machine (other node) is reachable on your LAN
--- ---
## Step 1 — Enable WSL2 and install Ubuntu ## Step 1 — Enable WSL2 and install Ubuntu
Open **PowerShell as Administrator** and run: Open **PowerShell as Administrator** and run:
```powershell ```powershell
wsl --install -d Ubuntu-24.04 wsl --install -d Ubuntu-24.04
``` ```
This installs WSL2 with Ubuntu 24.04. Reboot when prompted. This installs WSL2 with Ubuntu 24.04. Reboot when prompted.
After reboot, Ubuntu starts and asks you to create a UNIX username/password. Choose anything convenient. After reboot, Ubuntu starts and asks you to create a UNIX username/password. Choose anything convenient.
Verify WSL version: Verify WSL version:
```powershell ```powershell
wsl -l -v wsl -l -v
``` ```
Output should show `VERSION 2`. Output should show `VERSION 2`.
--- ---
## Step 2 — Install NVIDIA GPU driver on Windows (NOT inside WSL) ## Step 2 — Install NVIDIA GPU driver on Windows (NOT inside WSL)
WSL2 CUDA passthrough works through the Windows host driver. **Do not install CUDA inside WSL2.** WSL2 CUDA passthrough works through the Windows host driver. **Do not install CUDA inside WSL2.**
1. Download the latest Game Ready or Studio driver for your GPU from https://www.nvidia.com/drivers 1. Download the latest Game Ready or Studio driver for your GPU from https://www.nvidia.com/drivers
2. Install on Windows normally (standard installer). 2. Install on Windows normally (standard installer).
3. Inside WSL2 (Ubuntu terminal), verify: 3. Inside WSL2 (Ubuntu terminal), verify:
```bash ```bash
nvidia-smi nvidia-smi
``` ```
Expected output: your GPU name, driver version, CUDA version. If this command fails, the Windows driver is too old — update it. Expected output: your GPU name, driver version, CUDA version. If this command fails, the Windows driver is too old — update it.
> **Note:** The `cuda-toolkit` package inside WSL2 is optional and only needed if you compile CUDA kernels. For inference with `torch`, the Windows host driver is sufficient. > **Note:** The `cuda-toolkit` package inside WSL2 is optional and only needed if you compile CUDA kernels. For inference with `torch`, the Windows host driver is sufficient.
--- ---
## Step 3 — Install Python 3.11+ inside WSL2 ## Step 3 — Install Python 3.11+ inside WSL2
Ubuntu 24.04 ships Python 3.12. Confirm: Ubuntu 24.04 ships Python 3.12. Confirm:
```bash ```bash
python3 --version python3 --version
``` ```
If it shows 3.10 or older: If it shows 3.10 or older:
```bash ```bash
sudo add-apt-repository ppa:deadsnakes/ppa sudo add-apt-repository ppa:deadsnakes/ppa
sudo apt update sudo apt update
sudo apt install python3.12 python3.12-venv python3.12-dev sudo apt install python3.12 python3.12-venv python3.12-dev
``` ```
Install pip: Install pip:
```bash ```bash
curl -sS https://bootstrap.pypa.io/get-pip.py | python3 curl -sS https://bootstrap.pypa.io/get-pip.py | python3
``` ```
--- ---
## Step 4 — Clone the repository ## Step 4 — Clone the repository
Inside WSL2: Inside WSL2:
```bash ```bash
# Store the repo in the Linux filesystem (faster I/O than /mnt/c) # Store the repo in the Linux filesystem (faster I/O than /mnt/c)
cd ~ cd ~
git clone https://github.com/YOUR_ORG/d-popov.com.git git clone https://github.com/YOUR_ORG/d-popov.com.git
cd d-popov.com/AI cd d-popov.com/AI
``` ```
--- ---
## Step 5 — Create a virtualenv and install meshnet-node ## Step 5 — Create a virtualenv and install meshnet-node
```bash ```bash
python3 -m venv .venv python3 -m venv .venv
source .venv/bin/activate source .venv/bin/activate
# Install node + PyTorch (CUDA build) # Install node + PyTorch (CUDA build)
pip install torch --index-url https://download.pytorch.org/whl/cu124 pip install torch --index-url https://download.pytorch.org/whl/cu124
pip install -e "packages/node[torch]" pip install -e "packages/node[torch]"
``` ```
Verify the install: Verify the install:
```bash ```bash
meshnet-node --help 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
## Step 6 — Pre-download the model shard with `'Qwen3_5MoeConfig' object has no attribute 'vocab_size'`). If you install
into an existing conda/miniforge env instead of a fresh venv, run
Download the model before starting the node so the startup process doesn't time out on the tracker side: `pip install -U transformers` there. The startup warning about
`flash-linear-attention` / `causal-conv1d` ("fast path is not available") is
```bash harmless on CPU — those are optional CUDA-only kernels.
python3 - <<'EOF'
from transformers import AutoConfig ---
AutoConfig.from_pretrained("microsoft/Phi-3-medium-128k-instruct")
EOF ## Step 6 — Pre-download the model shard
```
Download the model before starting the node so the startup process doesn't time out on the tracker side:
For the full model weights (needed at runtime), `transformers` downloads them automatically on first `meshnet-node` start. If you want to pre-fetch:
```bash
```bash python3 - <<'EOF'
python3 -c " from transformers import AutoConfig
from transformers import AutoModelForCausalLM AutoConfig.from_pretrained("microsoft/Phi-3-medium-128k-instruct")
AutoModelForCausalLM.from_pretrained('microsoft/Phi-3-medium-128k-instruct', device_map='cpu') EOF
" ```
```
For the full model weights (needed at runtime), `transformers` downloads them automatically on first `meshnet-node` start. If you want to pre-fetch:
This can take 1030 minutes on first run.
```bash
--- python3 -c "
from transformers import AutoModelForCausalLM
## Step 7 — Expose the node port to your LAN AutoModelForCausalLM.from_pretrained('microsoft/Phi-3-medium-128k-instruct', device_map='cpu')
"
WSL2 runs behind a NAT with a virtual IP (typically `172.x.x.x`). Your LAN sees the Windows host IP. You need to forward the node port. ```
**Option A — Windows port proxy (recommended for simple setups):** This can take 1030 minutes on first run.
In **PowerShell as Administrator**: ---
```powershell ## Step 7 — Expose the node port to your LAN
# Get the current WSL2 IP (changes on each WSL restart)
$wslIp = (wsl hostname -I).Trim() WSL2 runs behind a NAT with a virtual IP (typically `172.x.x.x`). Your LAN sees the Windows host IP. You need to forward the node port.
# Forward Windows host port 8001 → WSL2 port 8001 **Option A — Windows port proxy (recommended for simple setups):**
netsh interface portproxy add v4tov4 `
listenport=8001 listenaddress=0.0.0.0 ` In **PowerShell as Administrator**:
connectport=8001 connectaddress=$wslIp
```powershell
# Allow inbound on Windows Firewall # Get the current WSL2 IP (changes on each WSL restart)
New-NetFirewallRule -DisplayName "meshnet-node" ` $wslIp = (wsl hostname -I).Trim()
-Direction Inbound -Protocol TCP -LocalPort 8001 -Action Allow
``` # Forward Windows host port 8001 → WSL2 port 8001
netsh interface portproxy add v4tov4 `
Verify: from the Linux machine, `curl http://WINDOWS_LAN_IP:8001/v1/health` should return a response once the node is running. listenport=8001 listenaddress=0.0.0.0 `
connectport=8001 connectaddress=$wslIp
**Redo this after every WSL2 restart** — the WSL2 IP changes.
# Allow inbound on Windows Firewall
**Option B — P2P relay (US-017, no port forwarding needed):** New-NetFirewallRule -DisplayName "meshnet-node" `
-Direction Inbound -Protocol TCP -LocalPort 8001 -Action Allow
Start a relay node on the Linux machine. The WSL2 node connects outbound through the relay. No firewall rules needed. See `docs/TWO_MACHINE_TEST.md` for details. ```
--- Verify: from the Linux machine, `curl http://WINDOWS_LAN_IP:8001/v1/health` should return a response once the node is running.
## Step 8 — Start the node **Redo this after every WSL2 restart** — the WSL2 IP changes.
Replace `192.168.1.10` with the actual LAN IP of the Linux machine running the tracker. **Option B — P2P relay (US-017, no port forwarding needed):**
Replace shard range with the complementary range to what the Linux node is serving.
Start a relay node on the Linux machine. The WSL2 node connects outbound through the relay. No firewall rules needed. See `docs/TWO_MACHINE_TEST.md` for details.
```bash
source .venv/bin/activate ---
meshnet-node \ ## Step 8 — Start the node
--model microsoft/Phi-3-medium-128k-instruct \
--quantization bf16 \ Replace `192.168.1.10` with the actual LAN IP of the Linux machine running the tracker.
--shard-start 20 --shard-end 39 \ Replace shard range with the complementary range to what the Linux node is serving.
--tracker http://192.168.1.10:8080 \
--port 8001 \ ```bash
--host 0.0.0.0 \ source .venv/bin/activate
--advertise-host WINDOWS_LAN_IP
``` meshnet-node \
--model microsoft/Phi-3-medium-128k-instruct \
The `--advertise-host` flag tells the tracker what IP the Linux machine should use to reach this node. Use your Windows machine's LAN IP (e.g. `192.168.1.20`), **not** the WSL2 internal IP. --quantization bf16 \
--shard-start 20 --shard-end 39 \
Expected startup output: --tracker http://192.168.1.10:8080 \
--port 8001 \
``` --host 0.0.0.0 \
Detecting hardware... --advertise-host WINDOWS_LAN_IP
GPU: NVIDIA GeForce RTX 3080 (10240 MB VRAM) ```
Loading wallet...
Wallet: 5K7r... The `--advertise-host` flag tells the tracker what IP the Linux machine should use to reach this node. Use your Windows machine's LAN IP (e.g. `192.168.1.20`), **not** the WSL2 internal IP.
Loading real PyTorch model shard...
Auto-detected 40 layers → shard 2039 Expected startup output:
================================
meshnet-node ready ```
Model ID: microsoft/Phi-3-medium-128k-instruct Detecting hardware...
Shard: layers 2039; 20 of 40 GPU: NVIDIA GeForce RTX 3080 (10240 MB VRAM)
Endpoint: http://192.168.1.20:8001 Loading wallet...
Hardware: CUDA Wallet: 5K7r...
================================ Loading real PyTorch model shard...
``` Auto-detected 40 layers → shard 2039
================================
--- meshnet-node ready
Model ID: microsoft/Phi-3-medium-128k-instruct
## Known issues Shard: layers 2039; 20 of 40
Endpoint: http://192.168.1.20:8001
- **WSL2 IP changes on restart.** Always re-run the `netsh` port-proxy command after restarting WSL2 or Windows. Hardware: CUDA
- **CUDA not visible in WSL2.** If `nvidia-smi` fails inside WSL2, update the Windows host GPU driver to ≥ 525.x. Installing CUDA inside WSL2 will not fix it. ================================
- **Model download is slow.** HuggingFace downloads happen over HTTPS. Pre-fetch the model before a timed test (see Step 6). ```
- **Port 8001 already in use.** Change `--port` to another value and update the firewall/portproxy rules accordingly.
- **`bf16` not supported on older GPUs.** Use `--quantization int8` on Turing (RTX 20xx) cards or earlier if bfloat16 ops fail. ---
## Known issues
- **WSL2 IP changes on restart.** Always re-run the `netsh` port-proxy command after restarting WSL2 or Windows.
- **CUDA not visible in WSL2.** If `nvidia-smi` fails inside WSL2, update the Windows host GPU driver to ≥ 525.x. Installing CUDA inside WSL2 will not fix it.
- **Model download is slow.** HuggingFace downloads happen over HTTPS. Pre-fetch the model before a timed test (see Step 6).
- **Port 8001 already in use.** Change `--port` to another value and update the firewall/portproxy rules accordingly.
- **`bf16` not supported on older GPUs.** Use `--quantization int8` on Turing (RTX 20xx) cards or earlier if bfloat16 ops fail.

Binary file not shown.

View File

@@ -286,7 +286,7 @@ class _GatewayHandler(http.server.BaseHTTPRequestHandler):
self._send_json(200, completion) self._send_json(200, completion)
def _proxy_to_head_worker(self, url: str, body_bytes: bytes) -> None: def _proxy_to_head_worker(self, url: str, body_bytes: bytes) -> None:
"""Forward a raw request body to a head worker and stream the response back.""" """Forward a raw request body to a head worker and relay SSE without buffering."""
target_url = f"{url}/v1/chat/completions" target_url = f"{url}/v1/chat/completions"
req = urllib.request.Request( req = urllib.request.Request(
target_url, target_url,
@@ -297,6 +297,19 @@ class _GatewayHandler(http.server.BaseHTTPRequestHandler):
try: try:
with urllib.request.urlopen(req, timeout=30.0) as r: with urllib.request.urlopen(req, timeout=30.0) as r:
content_type = r.headers.get("Content-Type", "application/json") content_type = r.headers.get("Content-Type", "application/json")
if "text/event-stream" in content_type:
self.send_response(r.status)
self.send_header("Content-Type", content_type)
self.send_header("Cache-Control", "no-cache")
self.send_header("X-Accel-Buffering", "no")
self.end_headers()
while True:
line = r.readline()
if not line:
break
self.wfile.write(line)
self.wfile.flush()
return
resp_body = r.read() resp_body = r.read()
status = r.status status = r.status
except urllib.error.HTTPError as exc: except urllib.error.HTTPError as exc:

File diff suppressed because it is too large Load Diff

View File

@@ -164,6 +164,9 @@ class RelayHttpBridge:
path = str(payload.get("path") or "/") path = str(payload.get("path") or "/")
headers = payload.get("headers") if isinstance(payload.get("headers"), dict) else {} headers = payload.get("headers") if isinstance(payload.get("headers"), dict) else {}
req_suffix = f" request_id={request_id}" if request_id else ""
print(f" [node] relay {method} {path}{req_suffix}", flush=True)
# body_base64 carries binary data (e.g. bfloat16 activation tensors) safely. # body_base64 carries binary data (e.g. bfloat16 activation tensors) safely.
# Fallback to text "body" for backward-compat with non-binary requests. # Fallback to text "body" for backward-compat with non-binary requests.
body_b64 = payload.get("body_base64") body_b64 = payload.get("body_base64")

View File

@@ -205,6 +205,21 @@ def _max_assignable_layers(
return min(total_layers, int((budget_bytes * 0.8) // layer_bytes)) return min(total_layers, int((budget_bytes * 0.8) // layer_bytes))
def _format_shard_label(
shard_start: int,
shard_end: int,
total_layers: int | None = None,
*,
model_name: str | None = None,
) -> str:
layer_count = shard_end - shard_start + 1
if isinstance(total_layers, int) and total_layers > 0:
return f"layers {shard_start}{shard_end} ({layer_count} of {total_layers})"
if model_name:
return f"layers {shard_start}{shard_end} ({model_name})"
return f"layers {shard_start}{shard_end}"
def _shard_budget_line( def _shard_budget_line(
memory_mb: int, memory_mb: int,
memory_source: str, memory_source: str,
@@ -734,11 +749,7 @@ def run_startup(
_node_start_time = time.monotonic() _node_start_time = time.monotonic()
actual_port = node.start() actual_port = node.start()
total_layers = getattr(getattr(node, "backend", None), "total_layers", None) total_layers = getattr(getattr(node, "backend", None), "total_layers", None)
if isinstance(total_layers, int) and total_layers > 0: shard_label = _format_shard_label(shard_start, shard_end, total_layers)
layer_count = shard_end - shard_start + 1
shard_label = f"layers {shard_start}{shard_end}; {layer_count} of {total_layers}"
else:
shard_label = f"layers {shard_start}{shard_end}"
public_host = advertise_host or (socket.getfqdn() if host == "0.0.0.0" else host) public_host = advertise_host or (socket.getfqdn() if host == "0.0.0.0" else host)
endpoint = f"http://{public_host}:{actual_port}" endpoint = f"http://{public_host}:{actual_port}"
local_base_url = f"http://127.0.0.1:{actual_port}" local_base_url = f"http://127.0.0.1:{actual_port}"
@@ -913,14 +924,17 @@ def run_startup(
tracker_node_id = _register_with_tracker( tracker_node_id = _register_with_tracker(
tracker_url, auto_reg_payload, node, _node_start_time, tracker_url, auto_reg_payload, node, _node_start_time,
) )
shard_count = assigned_shard_end - assigned_shard_start + 1 shard_label = _format_shard_label(
assigned_shard_start,
assigned_shard_end,
assigned_num_layers,
)
print( print(
f"\n{'=' * 32}\n" f"\n{'=' * 32}\n"
f"meshnet-node ready (auto-joined)\n" f"meshnet-node ready (auto-joined)\n"
f" Wallet: {address}\n" f" Wallet: {address}\n"
f" Model ID: {assigned_hf_repo}\n" f" Model ID: {assigned_hf_repo}\n"
f" Shard: layers {assigned_shard_start}{assigned_shard_end} " f" Shard: {shard_label}\n"
f"({shard_count} of {assigned_num_layers})\n"
f" {_shard_budget_line(memory_budget_mb, memory_budget_source, assigned_num_layers, quantization)}\n" f" {_shard_budget_line(memory_budget_mb, memory_budget_source, assigned_num_layers, quantization)}\n"
f" Quantization: {quantization}\n" f" Quantization: {quantization}\n"
f" Endpoint: {endpoint}\n" f" Endpoint: {endpoint}\n"
@@ -967,13 +981,35 @@ def run_startup(
peers: list[dict] = assignment.get("peers", []) peers: list[dict] = assignment.get("peers", [])
model_sources: list[dict] = [] if tracker_source_disabled else assignment.get("model_sources", []) model_sources: list[dict] = [] if tracker_source_disabled else assignment.get("model_sources", [])
assignment_bytes_per_layer = _assignment_bytes_per_layer(assignment, quantization) assignment_bytes_per_layer = _assignment_bytes_per_layer(assignment, quantization)
model_layers_end = assignment.get("model_layers_end")
assigned_total_layers = (
int(model_layers_end) + 1
if model_layers_end is not None
else None
)
shard_label = _format_shard_label(
shard_start,
shard_end,
assigned_total_layers,
model_name=assigned_model,
)
if user_pinned_shard: if user_pinned_shard:
print( shard_label = f"{shard_label} (pinned)"
f" Shard: layers {shard_start}-{shard_end} of {assigned_model} (pinned)", if user_pinned_shard and assigned_total_layers and assignment_bytes_per_layer:
flush=True, pinned_layers = shard_end - shard_start + 1
max_layers = _max_assignable_layers(
memory_budget_mb,
assigned_total_layers,
assignment_bytes_per_layer,
) )
else: if pinned_layers > max_layers:
print(f" Shard: layers {shard_start}-{shard_end} of {assigned_model}", flush=True) raise ValueError(
f"Pinned shard layers {shard_start}{shard_end} ({pinned_layers} layers) exceed "
f"the {memory_budget_mb / 1024:.1f} GB {memory_budget_source} budget "
f"(fits up to {max_layers}/{assigned_total_layers} layers at bfloat16). "
"Drop --shard-start/--shard-end to let the tracker auto-assign, or pin a smaller range."
)
print(f" Shard: {shard_label}", flush=True)
# 4. Download shard # 4. Download shard
print("Downloading shard...", flush=True) print("Downloading shard...", flush=True)
@@ -998,7 +1034,77 @@ def run_startup(
) )
print(f" Cached at: {shard_path}", flush=True) print(f" Cached at: {shard_path}", flush=True)
# 5. Start HTTP server # 5. Start HTTP server — real HF weights use TorchNodeServer; stub-model stays stub.
_node_start_time = time.monotonic()
if hf_repo and assigned_model != "stub-model":
print("Loading real PyTorch model shard...", flush=True)
node = TorchNodeServer(
host=host,
port=port,
model_id=hf_repo,
shard_start=shard_start,
shard_end=shard_end,
quantization=quantization,
tracker_url=tracker_url,
route_timeout=route_timeout,
cache_dir=shard_path,
debug=debug,
max_loaded_shards=max_loaded_shards,
)
actual_port = node.start()
total_layers = getattr(getattr(node, "backend", None), "total_layers", None) or assigned_total_layers
shard_label = _format_shard_label(shard_start, shard_end, total_layers, model_name=assigned_model)
if user_pinned_shard:
shard_label = f"{shard_label} (pinned)"
public_host = advertise_host or (socket.getfqdn() if host == "0.0.0.0" else host)
endpoint = f"http://{public_host}:{actual_port}"
local_base_url = f"http://127.0.0.1:{actual_port}"
relay_bridge, relay_fields = _start_relay_bridge_if_available(
tracker_url,
address,
local_base_url,
endpoint,
relay_url=relay_url,
)
_attach_relay_bridge(node, relay_bridge)
reg_payload = {
"endpoint": endpoint,
"model": assigned_model,
"hf_repo": hf_repo,
"num_layers": total_layers,
"shard_start": shard_start,
"shard_end": shard_end,
"downloaded_models": downloaded_models,
"hardware_profile": hw,
"wallet_address": address,
"quantization": quantization,
"score": 1.0,
"tracker_mode": (shard_start == 0),
"managed_assignment": not user_pinned_shard,
"model_metadata": model_metadata_for(hf_repo, total_layers, cache_dir=shard_path),
**registration_capabilities,
**relay_fields,
}
tracker_node_id = _register_with_tracker(
tracker_url, reg_payload, node, _node_start_time,
)
print(
f"\n{'=' * 32}\n"
f"meshnet-node ready\n"
f" Wallet: {address}\n"
f" Model ID: {hf_repo}\n"
f" Shard: {shard_label}\n"
f" {_shard_budget_line(memory_budget_mb, memory_budget_source, total_layers, quantization, bytes_per_layer=assignment_bytes_per_layer)}\n"
f" Quantization: {quantization}\n"
f" Endpoint: {endpoint}\n"
f" Node ID: {tracker_node_id or 'unregistered'}\n"
f" Hardware: {_hardware_label(device, gpu_name)}\n"
f" Benchmark: {bench_tps:,.0f} (throughput index)\n"
f"{'=' * 32}",
flush=True,
)
return node
is_last = shard_end >= assignment.get("model_layers_end", shard_end) is_last = shard_end >= assignment.get("model_layers_end", shard_end)
node = StubNodeServer( node = StubNodeServer(
host=host, host=host,
@@ -1009,7 +1115,6 @@ def run_startup(
model=assigned_model, model=assigned_model,
shard_path=shard_path, shard_path=shard_path,
) )
_node_start_time = time.monotonic()
actual_port = node.start() actual_port = node.start()
public_host = advertise_host or (socket.getfqdn() if host == "0.0.0.0" else host) public_host = advertise_host or (socket.getfqdn() if host == "0.0.0.0" else host)
endpoint = f"http://{public_host}:{actual_port}" endpoint = f"http://{public_host}:{actual_port}"
@@ -1055,12 +1160,18 @@ def run_startup(
hw_str = device.upper() hw_str = device.upper()
if gpu_name: if gpu_name:
hw_str += f" ({gpu_name}, {vram_mb / 1024:.1f} GB)" hw_str += f" ({gpu_name}, {vram_mb / 1024:.1f} GB)"
shard_label = _format_shard_label(
shard_start,
shard_end,
assigned_total_layers,
model_name=assigned_model,
)
print( print(
f"\n{'=' * 32}\n" f"\n{'=' * 32}\n"
f"meshnet-node ready\n" f"meshnet-node ready\n"
f" Wallet: {address}\n" f" Wallet: {address}\n"
f" Shard: layers {shard_start}-{shard_end} ({assigned_model})\n" f" Shard: {shard_label}\n"
f" {_shard_budget_line(memory_budget_mb, memory_budget_source, assignment.get('model_layers_end', shard_end) + 1, quantization, bytes_per_layer=assignment_bytes_per_layer)}\n" f" {_shard_budget_line(memory_budget_mb, memory_budget_source, assigned_total_layers, quantization, bytes_per_layer=assignment_bytes_per_layer)}\n"
f" Endpoint: {endpoint}\n" f" Endpoint: {endpoint}\n"
f" Node ID: {node_id}\n" f" Node ID: {node_id}\n"
f" Hardware: {hw_str}\n" f" Hardware: {hw_str}\n"

View File

@@ -113,6 +113,10 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler):
def log_message(self, fmt, *args): # noqa: suppress request logs in tests def log_message(self, fmt, *args): # noqa: suppress request logs in tests
pass pass
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 do_POST(self): def do_POST(self):
server: _TorchHTTPServer = self.server # type: ignore[assignment] server: _TorchHTTPServer = self.server # type: ignore[assignment]
if self.path == "/forward": if self.path == "/forward":
@@ -199,8 +203,18 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler):
return return
server.forward_chunk_count += 1 server.forward_chunk_count += 1
if int(self.headers.get("X-Meshnet-Hop-Index", "0")) > 0: hop_index = int(self.headers.get("X-Meshnet-Hop-Index", "0"))
if hop_index > 0:
server.received_activations = True 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_header = self.headers.get("X-Meshnet-Start-Layer")
start_layer = int(start_layer_header) if start_layer_header else None start_layer = int(start_layer_header) if start_layer_header else None
@@ -307,24 +321,57 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler):
if backend is None or not backend.is_head: if backend is None or not backend.is_head:
self._send_json(400, {"error": "model not loaded on this node"}) self._send_json(400, {"error": "model not loaded on this node"})
return return
max_tokens = int(body.get("max_tokens") or body.get("max_new_tokens") or 256) max_tokens = int(body.get("max_tokens") or body.get("max_new_tokens") or 5120)
temperature = float(body.get("temperature") or 1.0) temperature = float(body.get("temperature") or 1.0)
top_p = float(body.get("top_p") or 1.0) top_p = float(body.get("top_p") or 1.0)
print(
f" [node] processing chat model={model_name!r} stream={stream} "
f"max_tokens={max_tokens}{self._request_log_suffix()}",
flush=True,
)
# Fast path: this node owns the complete model — use HF generate() with KV cache. # Fast path: this node owns the complete model — use HF generate() with KV cache.
# Avoids the single-token-per-forward-pass limitation of the distributed path. # Avoids the single-token-per-forward-pass limitation of the distributed path.
if backend.is_head and backend.is_tail: if backend.is_head and backend.is_tail:
gen_started = time.monotonic()
try: try:
if stream: if stream:
self._stream_openai_response( token_count = 0
backend.generate_text_streaming(messages, max_tokens, temperature, top_p),
model_name, def _counting_stream():
nonlocal token_count
for token_text in backend.generate_text_streaming(
messages, max_tokens, temperature, top_p,
):
if token_text:
token_count += 1
yield token_text
self._stream_openai_response(_counting_stream(), model_name)
print(
f" [node] chat complete (stream) tokens={token_count} "
f"elapsed_s={time.monotonic() - gen_started:.1f}{self._request_log_suffix()}",
flush=True,
) )
else: else:
text = backend.generate_text(messages, max_tokens, temperature, top_p) text = backend.generate_text(messages, max_tokens, temperature, top_p)
completion_tokens = _backend_token_count(
backend, "count_text_tokens", text, fallback=len(text.split()) or 1,
)
print(
f" [node] chat complete tokens={completion_tokens} "
f"elapsed_s={time.monotonic() - gen_started:.1f}{self._request_log_suffix()}",
flush=True,
)
self._send_openai_response(text, model_name, False, messages, backend=backend) self._send_openai_response(text, model_name, False, messages, backend=backend)
except Exception as exc: except Exception as exc:
self._record_failed_request() self._record_failed_request()
print(
f" [node] chat failed after {time.monotonic() - gen_started:.1f}s: {exc}"
f"{self._request_log_suffix()}",
flush=True,
)
self._send_json(500, {"error": f"generation failed: {exc}"}) self._send_json(500, {"error": f"generation failed: {exc}"})
return return
@@ -368,7 +415,11 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler):
if stream: if stream:
stream_emit = self._start_openai_stream(model_name) stream_emit = self._start_openai_stream(model_name)
for _ in range(max_tokens): _GENERATION_LOG_INTERVAL = 5.0
gen_started = time.monotonic()
last_gen_log = gen_started
for step in range(max_tokens):
try: try:
payload = backend.encode_prompt(current_text) payload = backend.encode_prompt(current_text)
except Exception as exc: except Exception as exc:
@@ -386,6 +437,21 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler):
if stream_emit is not None: if stream_emit is not None:
stream_emit(token_str) stream_emit(token_str)
current_text = current_text + token_str current_text = current_text + token_str
now = time.monotonic()
if step == 0 or now - last_gen_log >= _GENERATION_LOG_INTERVAL:
print(
f" [node] generating step={step + 1}/{max_tokens} "
f"tokens={len(generated)} elapsed_s={now - gen_started:.1f}",
flush=True,
)
last_gen_log = now
if generated:
print(
f" [node] generation complete tokens={len(generated)} "
f"elapsed_s={time.monotonic() - gen_started:.1f}",
flush=True,
)
result_text = "".join(generated) result_text = "".join(generated)
if stream_emit is not None: if stream_emit is not None:

View File

@@ -1,33 +1,33 @@
[build-system] [build-system]
requires = ["setuptools>=64"] requires = ["setuptools>=64"]
build-backend = "setuptools.build_meta" build-backend = "setuptools.build_meta"
[project] [project]
name = "meshnet-node" name = "meshnet-node"
version = "0.1.0" version = "0.1.0"
description = "Distributed Inference Network node client" description = "Distributed Inference Network node client"
requires-python = ">=3.10" requires-python = ">=3.10"
dependencies = [ dependencies = [
"cryptography>=41", "cryptography>=41",
"huggingface-hub>=0.20", "huggingface-hub>=0.20",
"accelerate>=0.28", "accelerate>=0.28",
"bitsandbytes>=0.43", "bitsandbytes>=0.43",
"rich>=13", "rich>=13",
"safetensors>=0.4", "safetensors>=0.4",
"torch>=2.1", "torch>=2.1",
"transformers>=4.39", "transformers>=5.12",
"websockets>=13", "websockets>=13",
"zstandard>=0.22", "zstandard>=0.22",
"kernels>=0.11.1,<0.16", "kernels>=0.11.1,<0.16",
] ]
[project.scripts] [project.scripts]
meshnet-node = "meshnet_node.cli:main" meshnet-node = "meshnet_node.cli:main"
[tool.setuptools.packages.find] [tool.setuptools.packages.find]
where = ["."] where = ["."]
include = ["meshnet_node*"] include = ["meshnet_node*"]
[tool.setuptools.package-data] [tool.setuptools.package-data]
meshnet_node = ["*.json"] meshnet_node = ["*.json"]

View File

@@ -8,7 +8,8 @@ regular user.
Mutations are append-only events with unique ids — the same replication Mutations are append-only events with unique ids — the same replication
model as ``BillingLedger`` — so accounts and API keys converge across the model as ``BillingLedger`` — so accounts and API keys converge across the
tracker hive via gossip, and every dashboard can serve registration/login. tracker hive via gossip, and every dashboard can serve registration/login.
Sessions are deliberately local to each tracker (bearer tokens in memory). Sessions are local to each tracker and persisted so dashboard cookies survive
tracker restarts.
""" """
from __future__ import annotations from __future__ import annotations
@@ -115,6 +116,8 @@ class AccountStore:
"account_id": account_id, "account_id": account_id,
"expires": time.time() + SESSION_TTL, "expires": time.time() + SESSION_TTL,
} }
self._dirty = True
self.save_to_db()
return token return token
def session_account(self, token: str | None) -> dict | None: def session_account(self, token: str | None) -> dict | None:
@@ -134,7 +137,9 @@ class AccountStore:
if not token: if not token:
return return
with self._lock: with self._lock:
self._sessions.pop(token, None) if self._sessions.pop(token, None) is not None:
self._dirty = True
self.save_to_db()
# ---- API keys ---- # ---- API keys ----
@@ -271,6 +276,10 @@ class AccountStore:
"CREATE TABLE IF NOT EXISTS account_events " "CREATE TABLE IF NOT EXISTS account_events "
"(event_id TEXT PRIMARY KEY, payload TEXT NOT NULL, ts REAL NOT NULL)" "(event_id TEXT PRIMARY KEY, payload TEXT NOT NULL, ts REAL NOT NULL)"
) )
con.execute(
"CREATE TABLE IF NOT EXISTS account_sessions "
"(token TEXT PRIMARY KEY, account_id TEXT NOT NULL, expires REAL NOT NULL)"
)
con.commit() con.commit()
con.close() con.close()
@@ -279,6 +288,10 @@ class AccountStore:
rows = con.execute( rows = con.execute(
"SELECT payload FROM account_events ORDER BY ts, event_id" "SELECT payload FROM account_events ORDER BY ts, event_id"
).fetchall() ).fetchall()
session_rows = con.execute(
"SELECT token, account_id, expires FROM account_sessions WHERE expires >= ?",
(time.time(),),
).fetchall()
con.close() con.close()
with self._lock: with self._lock:
for (payload,) in rows: for (payload,) in rows:
@@ -288,6 +301,11 @@ class AccountStore:
continue continue
if event.get("id") not in self._seen_event_ids: if event.get("id") not in self._seen_event_ids:
self._apply_locked(event) self._apply_locked(event)
self._sessions = {
token: {"account_id": account_id, "expires": float(expires)}
for token, account_id, expires in session_rows
if account_id in self._accounts
}
self._dirty = False self._dirty = False
def save_to_db(self) -> None: def save_to_db(self) -> None:
@@ -297,11 +315,21 @@ class AccountStore:
if not self._dirty: if not self._dirty:
return return
events = list(self._event_log) events = list(self._event_log)
sessions = [
(token, session["account_id"], float(session["expires"]))
for token, session in self._sessions.items()
if session["expires"] >= time.time()
]
self._dirty = False self._dirty = False
con = sqlite3.connect(self._db_path) # type: ignore[arg-type] con = sqlite3.connect(self._db_path) # type: ignore[arg-type]
con.executemany( con.executemany(
"INSERT OR IGNORE INTO account_events (event_id, payload, ts) VALUES (?, ?, ?)", "INSERT OR IGNORE INTO account_events (event_id, payload, ts) VALUES (?, ?, ?)",
[(e["id"], json.dumps(e), float(e.get("ts", 0.0))) for e in events], [(e["id"], json.dumps(e), float(e.get("ts", 0.0))) for e in events],
) )
con.execute("DELETE FROM account_sessions")
con.executemany(
"INSERT INTO account_sessions (token, account_id, expires) VALUES (?, ?, ?)",
sessions,
)
con.commit() con.commit()
con.close() con.close()

View File

@@ -8,8 +8,14 @@ from pathlib import Path
from .accounts import DEFAULT_ACCOUNTS_DB_PATH from .accounts import DEFAULT_ACCOUNTS_DB_PATH
from .billing import DEFAULT_BILLING_DB_PATH from .billing import DEFAULT_BILLING_DB_PATH
from .hf_pricing import DEFAULT_HF_PRICING_LOG_DB_PATH from .hf_pricing import DEFAULT_HF_PRICING_LOG_DB_PATH
from .server import ( from .logging_setup import (
DEFAULT_LOG_BACKUP_COUNT,
DEFAULT_LOG_DIR,
DEFAULT_LOG_MAX_BYTES,
configure_tracker_file_logging,
)
from .server import (
DEFAULT_CALLER_CREDIT_USDT, DEFAULT_CALLER_CREDIT_USDT,
DEFAULT_DEVNET_TOPUP_USDT, DEFAULT_DEVNET_TOPUP_USDT,
TrackerServer, TrackerServer,
@@ -255,12 +261,40 @@ def main() -> None:
default=86400.0, default=86400.0,
help="Seconds between dynamic pricing refresh passes (default: daily)", help="Seconds between dynamic pricing refresh passes (default: daily)",
) )
common.add_argument( common.add_argument(
"--models-dir", "--models-dir",
default=None, default=None,
metavar="PATH", metavar="PATH",
help="Local HuggingFace snapshot root advertised as tracker model-file source (default: MESHNET_MODELS_DIR)", help="Local HuggingFace snapshot root advertised as tracker model-file source (default: MESHNET_MODELS_DIR)",
) )
common.add_argument(
"--log-dir",
default=DEFAULT_LOG_DIR,
metavar="PATH",
help=(
"Directory for rotating tracker logs "
f"(default: {DEFAULT_LOG_DIR}; files: info.log, warning.log, error.log)"
),
)
common.add_argument(
"--log-max-bytes",
type=int,
default=DEFAULT_LOG_MAX_BYTES,
metavar="BYTES",
help=f"Rotate each tracker log file after this many bytes (default: {DEFAULT_LOG_MAX_BYTES})",
)
common.add_argument(
"--log-backup-count",
type=int,
default=DEFAULT_LOG_BACKUP_COUNT,
metavar="N",
help=f"Number of rotated tracker log files to keep per level (default: {DEFAULT_LOG_BACKUP_COUNT})",
)
common.add_argument(
"--no-file-logs",
action="store_true",
help="Disable rotating tracker log files and only write to the terminal",
)
parser = argparse.ArgumentParser( parser = argparse.ArgumentParser(
prog="meshnet-tracker", prog="meshnet-tracker",
@@ -273,8 +307,15 @@ def main() -> None:
args = parser.parse_args() args = parser.parse_args()
if args.command in {None, "start"}: if args.command in {None, "start"}:
cluster_peers = [u.strip() for u in args.cluster_peers.split(",") if u.strip()] if not args.no_file_logs:
log_dir = configure_tracker_file_logging(
args.log_dir,
max_bytes=args.log_max_bytes,
backup_count=args.log_backup_count,
)
print(f"meshnet-tracker logs: {log_dir}", flush=True)
cluster_peers = [u.strip() for u in args.cluster_peers.split(",") if u.strip()]
relay_url = args.relay_url or derive_relay_url_from_public_tracker_url(args.self_url) relay_url = args.relay_url or derive_relay_url_from_public_tracker_url(args.self_url)
treasury = None treasury = None
if args.solana_rpc_url and args.usdt_mint and args.treasury_keypair: if args.solana_rpc_url and args.usdt_mint and args.treasury_keypair:

View File

@@ -5,17 +5,24 @@
<meta name="viewport" content="width=device-width, initial-scale=1"> <meta name="viewport" content="width=device-width, initial-scale=1">
<title>meshnet tracker</title> <title>meshnet tracker</title>
<style> <style>
:root { --bg:#0d1117; --panel:#161b22; --border:#30363d; --fg:#c9d1d9; :root { --bg:#0d1117; --panel:#161b22; --border:#30363d; --fg:#e6edf3;
--dim:#8b949e; --accent:#58a6ff; --ok:#3fb950; --bad:#f85149; --warn:#d29922; } --dim:#8b949e; --accent:#58a6ff; --ok:#3fb950; --bad:#f85149; --warn:#d29922;
--chat-input-bg:#21262d; --chat-user-bg:#1f4788; --chat-user-border:#388bfd; }
* { box-sizing:border-box; } * { box-sizing:border-box; }
html, body { height:100%; }
body { margin:0; background:var(--bg); color:var(--fg); body { margin:0; background:var(--bg); color:var(--fg);
font:13px/1.5 ui-monospace,SFMono-Regular,Menlo,Consolas,monospace; } font:13px/1.5 ui-monospace,SFMono-Regular,Menlo,Consolas,monospace; }
body.chat-tab-active { overflow:hidden; height:100dvh; display:flex; flex-direction:column; }
header { display:flex; align-items:baseline; gap:14px; padding:14px 20px; header { display:flex; align-items:baseline; gap:14px; padding:14px 20px;
border-bottom:1px solid var(--border); } border-bottom:1px solid var(--border); flex-shrink:0; }
header h1 { font-size:16px; margin:0; color:var(--accent); } header h1 { font-size:16px; margin:0; color:var(--accent); }
header .meta { color:var(--dim); font-size:12px; } header .meta { color:var(--dim); font-size:12px; }
main { display:grid; grid-template-columns:repeat(auto-fit,minmax(340px,1fr)); main { display:grid; grid-template-columns:repeat(auto-fit,minmax(340px,1fr));
gap:14px; padding:14px 20px; } gap:14px; padding:14px 20px; }
body.chat-tab-active main {
flex:1; min-height:0; display:flex; flex-direction:column;
padding:0; gap:0; overflow:hidden;
}
section { background:var(--panel); border:1px solid var(--border); section { background:var(--panel); border:1px solid var(--border);
border-radius:8px; padding:12px 14px; min-height:80px; } border-radius:8px; padding:12px 14px; min-height:80px; }
section h2 { margin:0 0 8px; font-size:12px; text-transform:uppercase; section h2 { margin:0 0 8px; font-size:12px; text-transform:uppercase;
@@ -53,27 +60,132 @@
.tabs { display:flex; gap:10px; margin-bottom:8px; } .tabs { display:flex; gap:10px; margin-bottom:8px; }
.tabs a { color:var(--dim); cursor:pointer; } .tabs a { color:var(--dim); cursor:pointer; }
.tabs a.active { color:var(--accent); border-bottom:1px solid var(--accent); } .tabs a.active { color:var(--accent); border-bottom:1px solid var(--accent); }
.dashboard-tabs { display:flex; gap:10px; padding:10px 20px 0; border-bottom:1px solid var(--border); } .dashboard-tabs { display:flex; gap:10px; padding:10px 20px 0; border-bottom:1px solid var(--border); flex-shrink:0; }
.dashboard-tabs button { border:0; border-bottom:1px solid transparent; border-radius:0; .dashboard-tabs button { border:0; border-bottom:1px solid transparent; border-radius:0;
background:transparent; color:var(--dim); padding:5px 0 8px; } background:transparent; color:var(--dim); padding:5px 0 8px; }
.dashboard-tabs button.active { color:var(--accent); border-bottom-color:var(--accent); } .dashboard-tabs button.active { color:var(--accent); border-bottom-color:var(--accent); }
.wide { grid-column:1 / -1; } .wide { grid-column:1 / -1; }
section[hidden] { display:none !important; } section[hidden] { display:none !important; }
.chat-shell { display:grid; grid-template-columns:minmax(0, 1.35fr) minmax(320px, 0.65fr); gap:12px; } section.chat-section {
.chat-pane { display:flex; flex-direction:column; gap:10px; min-width:0; } padding:0; border:0; border-radius:0; background:var(--bg); min-height:0;
.chat-panel { background:var(--bg); border:1px solid var(--border); border-radius:6px; padding:10px; } }
.chat-controls { display:flex; gap:10px; align-items:end; flex-wrap:wrap; } body.chat-tab-active section.chat-section {
.chat-controls label { display:flex; flex-direction:column; gap:4px; color:var(--dim); } flex:1; display:flex !important; flex-direction:column; min-height:0;
.chat-controls select { min-width:220px; } }
.chat-history { display:flex; flex-direction:column; gap:8px; min-height:220px; max-height:420px; overflow:auto; } .chat-app {
.chat-message { border:1px solid #21262d; border-radius:6px; padding:8px 10px; background:#10151d; } display:grid; grid-template-columns:260px minmax(0, 1fr); gap:0;
.chat-role { color:var(--dim); font-size:11px; text-transform:uppercase; letter-spacing:.06em; margin-bottom:4px; } flex:1; min-height:0; overflow:hidden; background:var(--bg);
.chat-role-user { color:var(--accent); } }
.chat-role-assistant { color:var(--ok); } .chat-sidebar {
.chat-role-error { color:var(--bad); } display:flex; flex-direction:column; min-height:0;
.chat-compose { display:flex; flex-direction:column; gap:8px; } border-right:1px solid var(--border); background:var(--panel);
.chat-compose textarea { min-height:112px; resize:vertical; width:100%; } }
.chat-status { color:var(--dim); font-size:12px; } .chat-new-btn {
margin:12px; width:calc(100% - 24px); text-align:left;
border:1px solid var(--border); border-radius:8px; padding:10px 12px;
background:transparent; color:var(--fg);
}
.chat-new-btn:hover { background:#10151d; border-color:var(--accent); }
.chat-session-list {
flex:1; overflow:auto; padding:0 8px 12px; display:flex; flex-direction:column; gap:2px;
}
.chat-session-list.empty-state {
justify-content:center; align-items:center; color:var(--dim); font-style:italic;
padding:24px 12px;
}
.chat-session-item {
position:relative; display:block; width:100%; text-align:left;
padding:10px 32px 10px 12px; border:1px solid transparent; border-radius:8px;
background:transparent; color:var(--fg); cursor:pointer;
}
.chat-session-item:hover { background:#10151d; }
.chat-session-item.active { background:#10151d; border-color:#30363d; }
.chat-session-title {
font-size:13px; white-space:nowrap; overflow:hidden; text-overflow:ellipsis;
}
.chat-session-meta {
margin-top:3px; font-size:11px; color:var(--dim);
white-space:nowrap; overflow:hidden; text-overflow:ellipsis;
}
.chat-session-delete {
position:absolute; top:50%; right:6px; transform:translateY(-50%);
padding:2px 6px; min-width:0; border:0; border-radius:4px;
background:transparent; color:var(--dim); line-height:1.2; opacity:0;
}
.chat-session-item:hover .chat-session-delete,
.chat-session-item.active .chat-session-delete { opacity:1; }
.chat-session-delete:hover { color:var(--bad); background:#1a1012; }
.chat-main { display:flex; flex-direction:column; min-height:0; min-width:0; color-scheme:dark; }
.chat-toolbar {
display:flex; gap:12px; align-items:center; flex-shrink:0;
padding:10px 16px; border-bottom:1px solid var(--border); background:var(--panel);
}
.chat-toolbar label {
display:flex; align-items:center; gap:8px; color:var(--dim); margin:0;
}
.chat-toolbar select {
min-width:220px; max-width:min(420px, 50vw);
color:var(--fg); background:var(--chat-input-bg); border:1px solid var(--border);
border-radius:6px; padding:6px 8px;
}
.chat-status { color:var(--dim); font-size:12px; margin-left:auto; }
.chat-messages {
flex:1; overflow:auto; padding:24px 16px; min-height:0;
background:var(--bg); color:var(--fg);
}
.chat-messages-inner {
max-width:768px; margin:0 auto; display:flex; flex-direction:column; gap:20px;
}
.chat-messages.empty .chat-messages-inner {
min-height:100%; justify-content:center; align-items:center;
color:var(--dim); font-size:14px;
}
.chat-row { display:flex; width:100%; }
.chat-row.user { justify-content:flex-end; }
.chat-row.assistant, .chat-row.error { justify-content:flex-start; }
.chat-bubble {
max-width:85%; padding:12px 14px; border-radius:16px; line-height:1.55;
white-space:pre-wrap; word-break:break-word; font-size:14px; color:var(--fg);
}
.chat-bubble.user {
background:var(--chat-user-bg); border:1px solid var(--chat-user-border);
border-bottom-right-radius:4px; color:#f0f6fc;
}
.chat-bubble.assistant {
background:var(--panel); border:1px solid var(--border);
border-bottom-left-radius:4px; max-width:100%; color:var(--fg);
}
.chat-bubble.error {
background:#1a1012; border:1px solid #5c2020; color:#ffb4b4; border-bottom-left-radius:4px;
}
.chat-compose-wrap {
flex-shrink:0; padding:12px 16px 16px; border-top:1px solid var(--border);
background:var(--panel);
}
.chat-compose {
display:flex; gap:8px; align-items:flex-end; max-width:768px; margin:0 auto;
padding:10px 12px; border:1px solid var(--border); border-radius:16px;
background:var(--chat-input-bg);
}
.chat-compose:focus-within {
border-color:var(--accent);
box-shadow:0 0 0 1px var(--accent);
}
.chat-compose textarea {
flex:1; min-height:24px; max-height:200px; resize:none; width:auto;
border:0; background:transparent; padding:4px 0; outline:none;
color:var(--fg); caret-color:var(--accent); font:inherit; font-size:14px; line-height:1.5;
}
.chat-compose textarea::placeholder { color:var(--dim); opacity:1; }
.chat-compose button {
flex-shrink:0; min-width:36px; height:36px; padding:0;
border-radius:8px; border:1px solid var(--chat-user-border);
background:var(--chat-user-bg); color:#f0f6fc;
}
.chat-compose button:hover:not(:disabled) {
border-color:var(--accent); background:#2563b8;
}
.chat-compose button:disabled { opacity:.45; cursor:not-allowed; }
.console { .console {
background:var(--bg); border:1px solid var(--border); border-radius:6px; background:var(--bg); border:1px solid var(--border); border-radius:6px;
min-height:160px; max-height:280px; overflow:auto; padding:7px 9px; min-height:160px; max-height:280px; overflow:auto; padding:7px 9px;
@@ -88,6 +200,9 @@
.status-processing { color:var(--accent); } .status-processing { color:var(--accent); }
.status-failed { color:var(--bad); } .status-failed { color:var(--bad); }
.status-complete { color:var(--ok); } .status-complete { color:var(--ok); }
.status-canceled { color:var(--dim); }
button.btn-cancel { color:var(--dim); padding:0 5px; min-width:1.4em; line-height:1.2; }
button.btn-cancel:hover { color:var(--bad); border-color:var(--bad); }
</style> </style>
</head> </head>
<body> <body>
@@ -108,27 +223,28 @@
<section data-tab="overview"><h2>Nodes &amp; coverage</h2><div id="nodes" class="empty">loading…</div></section> <section data-tab="overview"><h2>Nodes &amp; coverage</h2><div id="nodes" class="empty">loading…</div></section>
<section data-tab="overview"><h2>Model usage (RPM)</h2><div id="stats" class="empty">loading…</div></section> <section data-tab="overview"><h2>Model usage (RPM)</h2><div id="stats" class="empty">loading…</div></section>
<section data-tab="overview" class="wide"><h2>Call wall</h2><div id="call-wall" class="empty">loading...</div></section> <section data-tab="overview" class="wide"><h2>Call wall</h2><div id="call-wall" class="empty">loading...</div></section>
<section data-tab="chat" class="wide"> <section data-tab="chat" class="wide chat-section">
<h2>Chat / inference</h2> <h2 style="display:none">Chat / inference</h2>
<div class="chat-shell"> <div class="chat-app">
<div class="chat-pane"> <aside class="chat-sidebar">
<div class="chat-panel chat-controls"> <button type="button" class="chat-new-btn" onclick="createNewChatSession()">+ New chat</button>
<div id="chat-session-list" class="chat-session-list empty-state">No chats yet</div>
</aside>
<div class="chat-main">
<div class="chat-toolbar">
<label>Model <label>Model
<select id="chat-model" onchange="selectChatModel(this.value)"></select> <select id="chat-model" onchange="selectChatModel(this.value)"></select>
</label> </label>
<button class="small" onclick="clearChatHistory()">clear history</button>
</div>
<div class="chat-panel chat-compose">
<textarea id="chat-prompt" placeholder="Ask a question or describe the task"></textarea>
<div class="form-row">
<button onclick="sendChat()" id="chat-send">Send</button>
</div>
</div>
</div>
<div class="chat-pane">
<div class="chat-panel">
<div id="chat-status" class="chat-status">select a model to start</div> <div id="chat-status" class="chat-status">select a model to start</div>
<div id="chat-history" class="chat-history empty">no messages yet</div> </div>
<div id="chat-history" class="chat-messages empty">
<div class="chat-messages-inner">Send a message to start this conversation.</div>
</div>
<div class="chat-compose-wrap">
<div class="chat-compose">
<textarea id="chat-prompt" placeholder="Message…" rows="1" aria-label="Message"></textarea>
<button type="button" onclick="sendChat()" id="chat-send" title="Send (Enter)"></button>
</div>
</div> </div>
</div> </div>
</div> </div>
@@ -158,7 +274,7 @@ async function fetchJson(path) {
const headers = {}; const headers = {};
const token = localStorage.getItem("meshnet_session"); const token = localStorage.getItem("meshnet_session");
if (token) headers["Authorization"] = "Bearer " + token; if (token) headers["Authorization"] = "Bearer " + token;
const r = await fetch(path, { headers }); const r = await fetch(path, { headers, credentials: "same-origin" });
if (!r.ok) return null; if (!r.ok) return null;
return await r.json(); return await r.json();
} catch { return null; } } catch { return null; }
@@ -350,6 +466,13 @@ function buildCallWallStates(events) {
rec.elapsed = f.elapsed_seconds; rec.elapsed = f.elapsed_seconds;
rec.stream = f.stream; rec.stream = f.stream;
rec.terminal = e; rec.terminal = e;
} else if (msg === "proxy canceled") {
rec.status = "canceled";
rec.model = rec.model || f.model || f.route_model || "?";
rec.tokens = f.tokens;
rec.tps = f.tokens_per_sec;
rec.elapsed = f.elapsed_seconds;
rec.terminal = e;
} else if (msg === "proxy failed" || msg === "direct proxy failed after relay") { } else if (msg === "proxy failed" || msg === "direct proxy failed after relay") {
rec.status = "failed"; rec.status = "failed";
rec.model = rec.model || f.model || f.route_model || "?"; rec.model = rec.model || f.model || f.route_model || "?";
@@ -379,7 +502,7 @@ function renderCallWall(consoleData, stats) {
const terminal = []; const terminal = [];
for (const rec of states.values()) { for (const rec of states.values()) {
if (rec.status === "pending" || rec.status === "processing") active.push(rec); if (rec.status === "pending" || rec.status === "processing") active.push(rec);
else if (rec.status === "complete" || rec.status === "failed") terminal.push(rec); else if (rec.status === "complete" || rec.status === "failed" || rec.status === "canceled") terminal.push(rec);
} }
active.sort((a, b) => (a.started || 0) - (b.started || 0)); active.sort((a, b) => (a.started || 0) - (b.started || 0));
terminal.sort((a, b) => (b.terminal && b.terminal.ts) - (a.terminal && a.terminal.ts)); terminal.sort((a, b) => (b.terminal && b.terminal.ts) - (a.terminal && a.terminal.ts));
@@ -401,10 +524,13 @@ function renderCallWall(consoleData, stats) {
`</div>`; `</div>`;
if (active.length) { if (active.length) {
html += table(["status", "age", "model", "request", "live tps", "tokens", "queue", "route / note"], active.map(rec => { const canCancelProxies = isAdmin || !isLoggedIn;
const headers = ["status", "age", "model", "request", "live tps", "tokens", "queue", "route / note"];
if (canCancelProxies) headers.push("");
html += table(headers, active.map(rec => {
const statusCls = rec.status === "pending" ? "status-pending" : "status-processing"; const statusCls = rec.status === "pending" ? "status-pending" : "status-processing";
const note = rec.warn || (rec.route ? short(String(rec.route), 28) : ""); const note = rec.warn || (rec.route ? short(String(rec.route), 28) : "");
return [ const row = [
`<span class="${statusCls}">${esc(rec.status)}</span>`, `<span class="${statusCls}">${esc(rec.status)}</span>`,
`<span class="num">${esc(callWallAgeSeconds(rec, nowSec).toFixed(1))}s</span>`, `<span class="num">${esc(callWallAgeSeconds(rec, nowSec).toFixed(1))}s</span>`,
esc(short(rec.model || "?", 28)), esc(short(rec.model || "?", 28)),
@@ -414,6 +540,12 @@ function renderCallWall(consoleData, stats) {
`<span class="num">${esc(String(callWallMaxQueue(rec)))}</span>`, `<span class="num">${esc(String(callWallMaxQueue(rec)))}</span>`,
esc(note), esc(note),
]; ];
if (canCancelProxies) {
row.push(
`<button type="button" class="small btn-cancel" data-cancel-request="${esc(rec.id)}" title="Cancel request">×</button>`,
);
}
return row;
})); }));
} else { } else {
html += '<div class="empty">no in-flight requests</div>'; html += '<div class="empty">no in-flight requests</div>';
@@ -422,10 +554,16 @@ function renderCallWall(consoleData, stats) {
const historyRows = terminal.slice(0, 40).map(rec => { const historyRows = terminal.slice(0, 40).map(rec => {
const e = rec.terminal || {}; const e = rec.terminal || {};
const f = e.fields || {}; const f = e.fields || {};
const statusCls = rec.status === "failed" ? "status-failed" : "status-complete"; const statusCls = rec.status === "failed"
? "status-failed"
: rec.status === "canceled"
? "status-canceled"
: "status-complete";
const detail = rec.status === "failed" const detail = rec.status === "failed"
? esc(short(rec.error || "?", 40)) ? esc(short(rec.error || "?", 40))
: (f.stream ? "stream" : "json"); : rec.status === "canceled"
? "canceled"
: (f.stream ? "stream" : "json");
return [ return [
new Date((e.ts || 0) * 1000).toLocaleTimeString(), new Date((e.ts || 0) * 1000).toLocaleTimeString(),
`<span class="${statusCls}">${esc(rec.status)}</span>`, `<span class="${statusCls}">${esc(rec.status)}</span>`,
@@ -437,7 +575,7 @@ function renderCallWall(consoleData, stats) {
detail, detail,
]; ];
}); });
html += '<div style="margin-top:8px"><b class="dim">recent completed / failed</b></div>'; html += '<div style="margin-top:8px"><b class="dim">recent completed / failed / canceled</b></div>';
html += historyRows.length html += historyRows.length
? table(["time", "status", "model", "request", "tps", "tokens", "sec", "detail"], historyRows) ? table(["time", "status", "model", "request", "tps", "tokens", "sec", "detail"], historyRows)
: '<div class="empty">no completed requests yet</div>'; : '<div class="empty">no completed requests yet</div>';
@@ -579,17 +717,214 @@ let lastStats = null;
let availableModels = []; let availableModels = [];
let chatHistory = []; let chatHistory = [];
let chatBusy = false; let chatBusy = false;
let chatSessions = [];
let activeChatSessionId = "";
let selectedChatModel = localStorage.getItem("meshnet_chat_model") || ""; let selectedChatModel = localStorage.getItem("meshnet_chat_model") || "";
const CHAT_SESSIONS_KEY = "meshnet_chat_sessions_v1";
const CHAT_ACTIVE_SESSION_KEY = "meshnet_chat_active_session_v1";
const CHAT_SESSIONS_LIMIT = 50;
function newChatSessionId() {
if (window.crypto && crypto.randomUUID) return crypto.randomUUID();
return "chat-" + Date.now().toString(36) + "-" + Math.random().toString(36).slice(2, 8);
}
function loadChatSessionsStore() {
try {
const raw = localStorage.getItem(CHAT_SESSIONS_KEY);
const parsed = raw ? JSON.parse(raw) : [];
return Array.isArray(parsed) ? parsed : [];
} catch {
return [];
}
}
function saveChatSessionsStore() {
localStorage.setItem(CHAT_SESSIONS_KEY, JSON.stringify(chatSessions));
if (activeChatSessionId) {
localStorage.setItem(CHAT_ACTIVE_SESSION_KEY, activeChatSessionId);
}
}
function chatSessionTitle(session) {
const firstUser = (session.messages || []).find(msg => msg.role === "user");
if (!firstUser || !firstUser.content) return "New chat";
const text = String(firstUser.content).trim().replace(/\s+/g, " ");
return text.length > 42 ? text.slice(0, 42) + "…" : text;
}
function formatSessionTime(iso) {
if (!iso) return "";
const date = new Date(iso);
if (Number.isNaN(date.getTime())) return "";
const now = new Date();
const sameDay = date.toDateString() === now.toDateString();
if (sameDay) return date.toLocaleTimeString([], { hour: "2-digit", minute: "2-digit" });
return date.toLocaleDateString([], { month: "short", day: "numeric" });
}
function getActiveChatSession() {
return chatSessions.find(session => session.id === activeChatSessionId) || null;
}
function persistActiveChatSession() {
const session = getActiveChatSession();
if (!session) return;
session.messages = chatHistory.slice();
session.model = selectedChatModel || session.model || "";
session.title = chatSessionTitle(session);
session.updatedAt = new Date().toISOString();
chatSessions.sort((a, b) => String(b.updatedAt).localeCompare(String(a.updatedAt)));
if (chatSessions.length > CHAT_SESSIONS_LIMIT) {
chatSessions = chatSessions.slice(0, CHAT_SESSIONS_LIMIT);
if (!chatSessions.some(item => item.id === activeChatSessionId)) {
activeChatSessionId = chatSessions[0].id;
chatHistory = chatSessions[0].messages.slice();
}
}
saveChatSessionsStore();
renderChatSessionList();
}
function clearChatPrompt() {
const promptEl = $("chat-prompt");
if (!promptEl) return;
promptEl.value = "";
promptEl.style.height = "auto";
}
function createNewChatSession() {
if (chatBusy) return;
const session = {
id: newChatSessionId(),
title: "New chat",
model: selectedChatModel || "",
messages: [],
createdAt: new Date().toISOString(),
updatedAt: new Date().toISOString(),
};
chatSessions.unshift(session);
activeChatSessionId = session.id;
chatHistory = [];
clearChatPrompt();
saveChatSessionsStore();
renderChatSessionList();
renderChatHistory();
renderChatAuthHint();
const promptEl = $("chat-prompt");
if (promptEl) promptEl.focus();
}
function selectChatSession(sessionId) {
if (chatBusy) return;
const session = chatSessions.find(item => item.id === sessionId);
if (!session) return;
if (sessionId === activeChatSessionId) return;
activeChatSessionId = session.id;
chatHistory = (session.messages || []).slice();
clearChatPrompt();
if (session.model) {
selectedChatModel = session.model;
localStorage.setItem("meshnet_chat_model", selectedChatModel);
const select = $("chat-model");
if (select) select.value = selectedChatModel;
}
localStorage.setItem(CHAT_ACTIVE_SESSION_KEY, activeChatSessionId);
renderChatSessionList();
renderChatHistory();
renderChatAuthHint();
}
function deleteChatSession(sessionId, event) {
if (event) {
event.preventDefault();
event.stopPropagation();
}
if (chatBusy) return;
const index = chatSessions.findIndex(item => item.id === sessionId);
if (index < 0) return;
chatSessions.splice(index, 1);
if (activeChatSessionId === sessionId) {
if (chatSessions.length) {
activeChatSessionId = chatSessions[0].id;
chatHistory = (chatSessions[0].messages || []).slice();
clearChatPrompt();
if (chatSessions[0].model) {
selectedChatModel = chatSessions[0].model;
localStorage.setItem("meshnet_chat_model", selectedChatModel);
}
} else {
saveChatSessionsStore();
createNewChatSession();
return;
}
}
saveChatSessionsStore();
renderChatSessionList();
renderChatHistory();
renderChatModels();
}
function initChatSessions() {
chatSessions = loadChatSessionsStore();
activeChatSessionId = localStorage.getItem(CHAT_ACTIVE_SESSION_KEY) || "";
const active = chatSessions.find(session => session.id === activeChatSessionId);
if (!active) {
if (chatSessions.length) {
activeChatSessionId = chatSessions[0].id;
chatHistory = (chatSessions[0].messages || []).slice();
if (chatSessions[0].model) selectedChatModel = chatSessions[0].model;
} else {
createNewChatSession();
return;
}
} else {
chatHistory = (active.messages || []).slice();
if (active.model) selectedChatModel = active.model;
}
renderChatSessionList();
renderChatHistory();
}
function renderChatSessionList() {
const list = $("chat-session-list");
if (!list) return;
if (!chatSessions.length) {
list.className = "chat-session-list empty-state";
list.innerHTML = "No chats yet";
return;
}
list.className = "chat-session-list";
list.innerHTML = chatSessions.map(session => {
const active = session.id === activeChatSessionId ? " active" : "";
const title = esc(chatSessionTitle(session));
const when = esc(formatSessionTime(session.updatedAt || session.createdAt));
const id = JSON.stringify(session.id);
return `<div class="chat-session-item${active}" role="button" tabindex="0"` +
` onclick="selectChatSession(${id})"` +
` onkeydown="if(event.key==='Enter'||event.key===' '){event.preventDefault();selectChatSession(${id});}">` +
`<div class="chat-session-title">${title}</div>` +
(when ? `<div class="chat-session-meta">${when}</div>` : "") +
`<button type="button" class="chat-session-delete" title="Delete chat"` +
` onclick="deleteChatSession(${id}, event)">×</button>` +
`</div>`;
}).join("");
}
function switchDashboardTab(name) { function switchDashboardTab(name) {
if (name === "admin" && !isAdmin) name = "overview"; if (name === "admin" && !isAdmin) name = "overview";
if (name === "billing" && !isLoggedIn) name = "overview"; if (name === "billing" && !isLoggedIn) name = "overview";
dashboardTab = name; dashboardTab = name;
document.body.classList.toggle("chat-tab-active", name === "chat");
updateSectionVisibility(); updateSectionVisibility();
for (const tabName of ["overview", "chat", "billing", "admin"]) { for (const tabName of ["overview", "chat", "billing", "admin"]) {
const button = $("tab-" + tabName); const button = $("tab-" + tabName);
if (button) button.classList.toggle("active", tabName === dashboardTab); if (button) button.classList.toggle("active", tabName === dashboardTab);
} }
if (name === "chat") {
const promptEl = $("chat-prompt");
if (promptEl) promptEl.focus();
}
} }
function updateSectionVisibility() { function updateSectionVisibility() {
@@ -607,18 +942,18 @@ function renderChatStatus(text) {
function renderChatHistory() { function renderChatHistory() {
const history = $("chat-history"); const history = $("chat-history");
if (!history) return;
if (!chatHistory.length) { if (!chatHistory.length) {
history.classList.add("empty"); history.className = "chat-messages empty";
history.innerHTML = "no messages yet"; history.innerHTML = '<div class="chat-messages-inner">Send a message to start this conversation.</div>';
return; return;
} }
history.classList.remove("empty"); history.className = "chat-messages";
history.innerHTML = chatHistory.map(msg => { const rows = chatHistory.map(msg => {
const roleClass = msg.role === "user" ? "chat-role-user" : msg.role === "assistant" ? "chat-role-assistant" : "chat-role-error"; const role = msg.role === "user" ? "user" : msg.role === "assistant" ? "assistant" : "error";
const label = msg.role === "user" ? "user" : msg.role === "assistant" ? "assistant" : "error"; return `<div class="chat-row ${role}"><div class="chat-bubble ${role}">${esc(msg.content)}</div></div>`;
const meta = msg.model ? ` <span class="dim">· ${esc(short(msg.model, 24))}</span>` : "";
return `<div class="chat-message"><div class="chat-role ${roleClass}">${label}${meta}</div><div>${esc(msg.content)}</div></div>`;
}).join(""); }).join("");
history.innerHTML = `<div class="chat-messages-inner">${rows}</div>`;
history.scrollTop = history.scrollHeight; history.scrollTop = history.scrollHeight;
} }
@@ -649,12 +984,13 @@ function renderChatModels() {
function selectChatModel(value) { function selectChatModel(value) {
selectedChatModel = value || ""; selectedChatModel = value || "";
localStorage.setItem("meshnet_chat_model", selectedChatModel); localStorage.setItem("meshnet_chat_model", selectedChatModel);
} const session = getActiveChatSession();
if (session) {
function clearChatHistory() { session.model = selectedChatModel;
chatHistory = []; session.updatedAt = new Date().toISOString();
renderChatHistory(); saveChatSessionsStore();
renderChatStatus("history cleared"); renderChatSessionList();
}
} }
function chatAuthToken() { function chatAuthToken() {
@@ -694,6 +1030,7 @@ async function apiCall(path, method, body, bearerToken) {
const r = await fetch(path, { const r = await fetch(path, {
method: method || "GET", method: method || "GET",
headers, headers,
credentials: "same-origin",
body: body ? JSON.stringify(body) : undefined, body: body ? JSON.stringify(body) : undefined,
}); });
const data = await r.json().catch(() => ({})); const data = await r.json().catch(() => ({}));
@@ -904,40 +1241,107 @@ async function sendChat() {
.map(msg => ({ role: msg.role, content: msg.content })), .map(msg => ({ role: msg.role, content: msg.content })),
{ role: "user", content: prompt }, { role: "user", content: prompt },
], ],
stream: false, stream: true,
max_tokens: 256, max_tokens: 15120,
}; };
chatBusy = true; chatBusy = true;
$("chat-send").disabled = true; $("chat-send").disabled = true;
promptEl.value = ""; promptEl.value = "";
promptEl.style.height = "auto";
chatHistory.push({ role: "user", content: prompt, model: selectedChatModel }); chatHistory.push({ role: "user", content: prompt, model: selectedChatModel });
const assistantMessage = { role: "assistant", content: "", model: selectedChatModel };
chatHistory.push(assistantMessage);
renderChatHistory(); renderChatHistory();
renderChatStatus("sending request…"); persistActiveChatSession();
const r = await apiCall("/v1/chat/completions", "POST", body, bearerToken); renderChatStatus("streaming response…");
chatBusy = false;
$("chat-send").disabled = false; try {
if (!r.ok) { const headers = { "Content-Type": "application/json" };
const error = r.data && r.data.error if (bearerToken) headers["Authorization"] = "Bearer " + bearerToken;
? (typeof r.data.error === "string" ? r.data.error : r.data.error.message || "request failed") const response = await fetch("/v1/chat/completions", {
: "request failed"; method: "POST",
chatHistory.push({ role: "error", content: error, model: selectedChatModel }); headers,
credentials: "same-origin",
body: JSON.stringify(body),
});
if (!response.ok) {
const data = await response.json().catch(() => ({}));
const error = data && data.error
? (typeof data.error === "string" ? data.error : data.error.message || "request failed")
: "request failed";
assistantMessage.role = "error";
assistantMessage.content = error;
renderChatHistory();
persistActiveChatSession();
renderChatStatus(error);
return;
}
if (!response.body) throw new Error("stream response body is missing");
const reader = response.body.getReader();
const decoder = new TextDecoder();
let buffer = "";
let receivedAny = false;
while (true) {
const { value, done } = await reader.read();
if (done) break;
buffer += decoder.decode(value, { stream: true });
const events = buffer.split("\n\n");
buffer = events.pop() || "";
for (const eventText of events) {
for (const line of eventText.split("\n")) {
if (!line.startsWith("data: ")) continue;
const data = line.slice(6).trim();
if (data === "[DONE]") {
buffer = "";
break;
}
try {
const chunk = JSON.parse(data);
const delta = chunk.choices && chunk.choices[0] && chunk.choices[0].delta;
if (delta && delta.content) {
assistantMessage.content += delta.content;
receivedAny = true;
renderChatHistory();
persistActiveChatSession();
}
} catch {
// Ignore malformed SSE keepalive/event lines.
}
}
}
}
if (!receivedAny) assistantMessage.content = "(empty response)";
renderChatHistory(); renderChatHistory();
renderChatStatus(error); persistActiveChatSession();
renderChatStatus("done");
} catch (err) {
assistantMessage.role = "error";
assistantMessage.content = err && err.message ? err.message : "request failed";
renderChatHistory();
persistActiveChatSession();
renderChatStatus(assistantMessage.content);
} finally {
chatBusy = false;
$("chat-send").disabled = false;
promptEl.focus(); promptEl.focus();
return;
} }
const reply = (r.data && r.data.choices && r.data.choices[0] && r.data.choices[0].message && r.data.choices[0].message.content) || ""; }
const usage = r.data && r.data.usage;
chatHistory.push({ function bindChatPromptShortcuts() {
role: "assistant", const promptEl = $("chat-prompt");
content: reply || "(empty response)", if (!promptEl || promptEl.dataset.bound === "1") return;
model: selectedChatModel, promptEl.dataset.bound = "1";
promptEl.addEventListener("keydown", event => {
if (event.key === "Enter" && !event.shiftKey) {
event.preventDefault();
sendChat();
}
});
promptEl.addEventListener("input", () => {
promptEl.style.height = "auto";
promptEl.style.height = Math.min(promptEl.scrollHeight, 200) + "px";
}); });
renderChatHistory();
renderChatStatus(usage
? `done: ${usage.total_tokens ?? "?"} tokens`
: "done");
promptEl.focus();
} }
async function renderAdminPanel() { async function renderAdminPanel() {
@@ -956,6 +1360,23 @@ async function renderAdminPanel() {
$("admin").innerHTML = table(["account", "role", "keys", "balance (USDT)", "created"], rows); $("admin").innerHTML = table(["account", "role", "keys", "balance (USDT)", "created"], rows);
} }
async function cancelProxyRequest(requestId) {
const r = await apiCall(
`/v1/proxy/requests/${encodeURIComponent(requestId)}/cancel`,
"POST",
{},
);
if (r.ok) refresh();
}
$("call-wall").addEventListener("click", (event) => {
const button = event.target.closest("[data-cancel-request]");
if (!button) return;
event.preventDefault();
const requestId = button.getAttribute("data-cancel-request");
if (requestId) cancelProxyRequest(requestId);
});
async function refresh() { async function refresh() {
$("self-url").textContent = location.host; $("self-url").textContent = location.host;
const [raft, map, stats, models, consoleData, adminData] = await Promise.all([ const [raft, map, stats, models, consoleData, adminData] = await Promise.all([
@@ -992,12 +1413,14 @@ async function refresh() {
$("refreshed").textContent = "refreshed " + new Date().toLocaleTimeString(); $("refreshed").textContent = "refreshed " + new Date().toLocaleTimeString();
} }
refresh(); refresh();
initChatSessions();
bindChatPromptShortcuts();
renderAccountPanel(); renderAccountPanel();
renderChatModels(); renderChatModels();
renderChatHistory(); renderChatHistory();
renderChatAuthHint(); renderChatAuthHint();
setInterval(refresh, 4000); setInterval(refresh, 4000);
setInterval(() => { if (sessionToken) renderAccountPanel(); }, 8000); setInterval(() => { if (sessionToken || isLoggedIn) renderAccountPanel(); }, 8000);
</script> </script>
</body> </body>
</html> </html>

View File

@@ -0,0 +1,99 @@
"""Rotating file logging for the tracker CLI."""
from __future__ import annotations
import logging
import sys
from logging.handlers import RotatingFileHandler
from pathlib import Path
from typing import TextIO
DEFAULT_LOG_DIR = "logs/tracker"
DEFAULT_LOG_MAX_BYTES = 10 * 1024 * 1024
DEFAULT_LOG_BACKUP_COUNT = 5
TRACKER_LOGGER_NAME = "meshnet.tracker"
class _ExactLevelFilter(logging.Filter):
def __init__(self, level: int) -> None:
super().__init__()
self._level = level
def filter(self, record: logging.LogRecord) -> bool:
return record.levelno == self._level
class _TeeStream:
def __init__(self, stream: TextIO, logger: logging.Logger, level: int) -> None:
self._stream = stream
self._logger = logger
self._level = level
self._buffer = ""
def write(self, text: str) -> int:
self._stream.write(text)
self._stream.flush()
self._buffer += text
while "\n" in self._buffer:
line, self._buffer = self._buffer.split("\n", 1)
line = line.rstrip()
if line:
self._logger.log(self._level, line)
return len(text)
def flush(self) -> None:
self._stream.flush()
line = self._buffer.rstrip()
if line:
self._logger.log(self._level, line)
self._buffer = ""
def isatty(self) -> bool:
return self._stream.isatty()
def _make_handler(path: Path, level: int, *, max_bytes: int, backup_count: int) -> RotatingFileHandler:
handler = RotatingFileHandler(
path,
maxBytes=max_bytes,
backupCount=backup_count,
encoding="utf-8",
)
handler.setLevel(level)
handler.addFilter(_ExactLevelFilter(level))
handler.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(message)s"))
return handler
def configure_tracker_file_logging(
log_dir: str | Path = DEFAULT_LOG_DIR,
*,
max_bytes: int = DEFAULT_LOG_MAX_BYTES,
backup_count: int = DEFAULT_LOG_BACKUP_COUNT,
tee_stdio: bool = True,
) -> Path:
"""Configure rotatable info/warning/error log files and return the directory."""
path = Path(log_dir).expanduser()
path.mkdir(parents=True, exist_ok=True)
logger = logging.getLogger(TRACKER_LOGGER_NAME)
logger.setLevel(logging.INFO)
logger.propagate = False
logger.handlers.clear()
logger.addHandler(_make_handler(path / "info.log", logging.INFO, max_bytes=max_bytes, backup_count=backup_count))
logger.addHandler(_make_handler(path / "warning.log", logging.WARNING, max_bytes=max_bytes, backup_count=backup_count))
logger.addHandler(_make_handler(path / "error.log", logging.ERROR, max_bytes=max_bytes, backup_count=backup_count))
if tee_stdio:
if not isinstance(sys.stdout, _TeeStream):
sys.stdout = _TeeStream(sys.stdout, logger, logging.INFO) # type: ignore[assignment]
if not isinstance(sys.stderr, _TeeStream):
sys.stderr = _TeeStream(sys.stderr, logger, logging.ERROR) # type: ignore[assignment]
return path
def tracker_logger() -> logging.Logger:
return logging.getLogger(TRACKER_LOGGER_NAME)

View File

@@ -21,11 +21,13 @@ HTTP API contract:
Response 400/404/503: {"error": str} Response 400/404/503: {"error": str}
""" """
import http.cookies
import http.server import http.server
import hashlib import hashlib
import itertools import itertools
import json import json
import os import os
import select
import socketserver import socketserver
import sqlite3 import sqlite3
import tarfile import tarfile
@@ -35,6 +37,7 @@ import urllib.parse
import urllib.request import urllib.request
import uuid import uuid
from collections import deque from collections import deque
from dataclasses import dataclass, field
from importlib.resources import files from importlib.resources import files
from pathlib import Path from pathlib import Path
from typing import Any from typing import Any
@@ -46,11 +49,14 @@ from .billing import DEFAULT_BILLING_DB_PATH, BillingLedger
from .calibration import DEFAULT_CALIBRATION_DB_PATH, ToplocCalibrationStore from .calibration import DEFAULT_CALIBRATION_DB_PATH, ToplocCalibrationStore
from .hf_pricing import DEFAULT_HF_PRICING_LOG_DB_PATH, HfPricingLog, refresh_preset_price from .hf_pricing import DEFAULT_HF_PRICING_LOG_DB_PATH, HfPricingLog, refresh_preset_price
from .gossip import NodeGossip from .gossip import NodeGossip
from .logging_setup import tracker_logger
from .model_files import files_for_layer_range, snapshot_dir_for_repo from .model_files import files_for_layer_range, snapshot_dir_for_repo
from .raft import RaftNode from .raft import RaftNode
_CONSOLE_LIMIT = 300 _CONSOLE_LIMIT = 300
_PROXY_PROGRESS_LOG_INTERVAL = 5.0
_SESSION_COOKIE_NAME = "meshnet_session"
def _preset_price_keys(name: str, preset: dict) -> set[str]: def _preset_price_keys(name: str, preset: dict) -> set[str]:
@@ -1413,6 +1419,10 @@ def _relay_http_request_frames(
headers: dict[str, str], headers: dict[str, str],
timeout: float = 310.0, timeout: float = 310.0,
idle_timeout: float = 120.0, idle_timeout: float = 120.0,
*,
cancel_event: threading.Event | None = None,
ws_holder: list[Any] | None = None,
ws_lock: threading.Lock | None = None,
): ):
"""Send an HTTP-shaped request through a relay RPC WebSocket, yielding """Send an HTTP-shaped request through a relay RPC WebSocket, yielding
response frames until a terminal one (US-036). response frames until a terminal one (US-036).
@@ -1430,6 +1440,14 @@ def _relay_http_request_frames(
deadline = time.monotonic() + timeout deadline = time.monotonic() + timeout
try: try:
with wsc.connect(relay_addr, open_timeout=10, close_timeout=5) as ws: with wsc.connect(relay_addr, open_timeout=10, close_timeout=5) as ws:
if ws_holder is not None:
if ws_lock is not None:
with ws_lock:
ws_holder.clear()
ws_holder.append(ws)
else:
ws_holder.clear()
ws_holder.append(ws)
ws.send(json.dumps({ ws.send(json.dumps({
"request_id": request_id, "request_id": request_id,
"method": "POST", "method": "POST",
@@ -1438,6 +1456,8 @@ def _relay_http_request_frames(
"body": body.decode(errors="replace"), "body": body.decode(errors="replace"),
})) }))
while True: while True:
if cancel_event is not None and cancel_event.is_set():
return
remaining = deadline - time.monotonic() remaining = deadline - time.monotonic()
if remaining <= 0: if remaining <= 0:
return return
@@ -1917,6 +1937,38 @@ def _api_key_from_headers(headers) -> str | None:
return auth.strip() or None return auth.strip() or None
def _session_token_from_headers(headers) -> str | None:
token = _api_key_from_headers(headers)
if token:
return token
cookie_header = headers.get("Cookie")
if not cookie_header:
return None
cookie = http.cookies.SimpleCookie()
try:
cookie.load(cookie_header)
except http.cookies.CookieError:
return None
morsel = cookie.get(_SESSION_COOKIE_NAME)
if morsel is None:
return None
return morsel.value.strip() or None
def _session_cookie_header(token: str | None) -> str:
cookie = http.cookies.SimpleCookie()
cookie[_SESSION_COOKIE_NAME] = token or ""
morsel = cookie[_SESSION_COOKIE_NAME]
morsel["path"] = "/"
morsel["httponly"] = True
morsel["samesite"] = "Lax"
if token:
morsel["max-age"] = str(int(7 * 86400))
else:
morsel["max-age"] = "0"
return morsel.OutputString()
def _usage_total_tokens(payload: dict) -> int | None: def _usage_total_tokens(payload: dict) -> int | None:
usage = payload.get("usage") usage = payload.get("usage")
if not isinstance(usage, dict): if not isinstance(usage, dict):
@@ -2076,7 +2128,22 @@ def _registration_ban_error(contracts: Any | None, wallet_address: str | None) -
return None return None
def _tracker_log(server: "_TrackerHTTPServer", level: str, message: str, **fields: Any) -> None: def _tracker_log(
server: "_TrackerHTTPServer",
level: str,
message: str,
*,
stdout: bool = True,
update_console_key: str | None = None,
**fields: Any,
) -> None:
log_level = {
"debug": 10,
"info": 20,
"warn": 30,
"warning": 30,
"error": 40,
}.get(level.lower(), 20)
event = { event = {
"ts": time.time(), "ts": time.time(),
"level": level, "level": level,
@@ -2088,10 +2155,93 @@ def _tracker_log(server: "_TrackerHTTPServer", level: str, message: str, **field
}, },
} }
with server.console_lock: with server.console_lock:
server.console_events.append(event) if update_console_key is not None:
updated = False
for existing in reversed(server.console_events):
if (
existing.get("message") == message
and existing.get("fields", {}).get("request_id") == update_console_key
):
existing["ts"] = event["ts"]
existing["fields"] = event["fields"]
updated = True
break
if not updated:
server.console_events.append(event)
else:
server.console_events.append(event)
extras = " ".join(f"{key}={value}" for key, value in event["fields"].items()) extras = " ".join(f"{key}={value}" for key, value in event["fields"].items())
suffix = f" {extras}" if extras else "" suffix = f" {extras}" if extras else ""
print(f"[tracker] {level}: {message}{suffix}", flush=True) tracker_logger().log(log_level, f"{message}{suffix}")
if stdout:
print(f"[tracker] {level}: {message}{suffix}", flush=True)
@dataclass
class _ActiveProxyContext:
request_id: str
cancel_event: threading.Event = field(default_factory=threading.Event)
upstream: Any | None = None
upstream_lock: threading.Lock = field(default_factory=threading.Lock)
relay_ws: Any | None = None
relay_ws_lock: threading.Lock = field(default_factory=threading.Lock)
def _register_active_proxy(server: "_TrackerHTTPServer", request_id: str) -> _ActiveProxyContext:
ctx = _ActiveProxyContext(request_id=request_id)
with server.active_proxies_lock:
server.active_proxies[request_id] = ctx
return ctx
def _unregister_active_proxy(server: "_TrackerHTTPServer", request_id: str) -> None:
with server.active_proxies_lock:
server.active_proxies.pop(request_id, None)
def _request_proxy_cancel(server: "_TrackerHTTPServer", request_id: str) -> bool:
with server.active_proxies_lock:
ctx = server.active_proxies.get(request_id)
if ctx is None:
return False
ctx.cancel_event.set()
def _close_resources() -> None:
with ctx.upstream_lock:
upstream = ctx.upstream
if upstream is not None:
try:
upstream.close()
except Exception:
pass
with ctx.relay_ws_lock:
relay_ws = ctx.relay_ws
if relay_ws is not None:
try:
relay_ws.close()
except Exception:
pass
threading.Thread(target=_close_resources, daemon=True).start()
return True
def _upstream_socket(upstream: Any) -> Any | None:
fp = getattr(upstream, "fp", None)
raw = getattr(fp, "raw", None) if fp is not None else None
return getattr(raw, "_sock", None) if raw is not None else None
def _set_upstream_read_timeout(upstream: Any, timeout: float | None) -> None:
sock = _upstream_socket(upstream)
if sock is not None:
sock.settimeout(timeout)
def _clear_proxy_progress_log_state(server: "_TrackerHTTPServer", request_id: str) -> None:
state = getattr(server, "_proxy_progress_log_state", None)
if state is not None:
state.pop(request_id, None)
def _tracker_log_proxy_progress( def _tracker_log_proxy_progress(
@@ -2108,10 +2258,21 @@ def _tracker_log_proxy_progress(
) -> None: ) -> None:
elapsed = time.monotonic() - started elapsed = time.monotonic() - started
effective_elapsed = max(elapsed, 1e-6) effective_elapsed = max(elapsed, 1e-6)
now = time.monotonic()
state = getattr(server, "_proxy_progress_log_state", None)
if state is None:
state = {}
server._proxy_progress_log_state = state
last_stdout = state.get(request_id)
stdout = last_stdout is None or (now - last_stdout) >= _PROXY_PROGRESS_LOG_INTERVAL
if stdout:
state[request_id] = now
_tracker_log( _tracker_log(
server, server,
"info", "info",
"proxy progress", "proxy progress",
stdout=stdout,
update_console_key=request_id,
request_id=request_id, request_id=request_id,
model=model, model=model,
route_model=route_model, route_model=route_model,
@@ -2209,17 +2370,21 @@ class _TrackerHTTPServer(socketserver.ThreadingMixIn, http.server.HTTPServer):
self.models_dir = models_dir self.models_dir = models_dir
self.console_events = deque(maxlen=_CONSOLE_LIMIT) self.console_events = deque(maxlen=_CONSOLE_LIMIT)
self.console_lock = threading.Lock() self.console_lock = threading.Lock()
self.active_proxies: dict[str, _ActiveProxyContext] = {}
self.active_proxies_lock = threading.Lock()
class _TrackerHandler(http.server.BaseHTTPRequestHandler): class _TrackerHandler(http.server.BaseHTTPRequestHandler):
def log_message(self, fmt, *args): # noqa: suppress request logs in tests def log_message(self, fmt, *args): # noqa: suppress request logs in tests
pass pass
def _send_json(self, status: int, data: dict) -> None: def _send_json(self, status: int, data: dict, headers: dict[str, str] | None = None) -> None:
body = json.dumps(data).encode() body = json.dumps(data).encode()
self.send_response(status) self.send_response(status)
self.send_header("Content-Type", "application/json") self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(body))) self.send_header("Content-Length", str(len(body)))
for name, value in (headers or {}).items():
self.send_header(name, value)
self.end_headers() self.end_headers()
try: try:
self.wfile.write(body) self.wfile.write(body)
@@ -2236,7 +2401,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
inference and wallet binding only, never operator endpoints. inference and wallet binding only, never operator endpoints.
""" """
server: _TrackerHTTPServer = self.server # type: ignore[assignment] server: _TrackerHTTPServer = self.server # type: ignore[assignment]
token = _api_key_from_headers(self.headers) token = _session_token_from_headers(self.headers)
if not token: if not token:
return None, None return None, None
if is_validator_token(token, server.validator_service_token): if is_validator_token(token, server.validator_service_token):
@@ -2364,6 +2529,17 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
if len(parts) == 5 and parts[1] == "v1" and parts[2] == "nodes" and parts[4] == "heartbeat": if len(parts) == 5 and parts[1] == "v1" and parts[2] == "nodes" and parts[4] == "heartbeat":
self._handle_heartbeat(parts[3]) self._handle_heartbeat(parts[3])
return return
# /v1/proxy/requests/<request_id>/cancel
if (
len(parts) == 6
and parts[1] == "v1"
and parts[2] == "proxy"
and parts[3] == "requests"
and parts[5] == "cancel"
and parts[4]
):
self._handle_proxy_request_cancel(urllib.parse.unquote(parts[4]))
return
self.send_response(404) self.send_response(404)
self.end_headers() self.end_headers()
@@ -2886,6 +3062,9 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
if inflight_recorded: if inflight_recorded:
_record_proxy_inflight(server, inflight_nodes, -1) _record_proxy_inflight(server, inflight_nodes, -1)
inflight_recorded = False inflight_recorded = False
_unregister_active_proxy(server, request_id)
proxy_ctx = _register_active_proxy(server, request_id)
_tracker_log( _tracker_log(
server, server,
@@ -2906,6 +3085,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
headers={ headers={
"Content-Type": "application/json", "Content-Type": "application/json",
"X-Meshnet-Route": downstream_urls, "X-Meshnet-Route": downstream_urls,
"X-Meshnet-Request-Id": request_id,
}, },
method="POST", method="POST",
) )
@@ -2917,6 +3097,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
relay_headers = { relay_headers = {
"Content-Type": "application/json", "Content-Type": "application/json",
"X-Meshnet-Route": downstream_urls, "X-Meshnet-Route": downstream_urls,
"X-Meshnet-Request-Id": request_id,
**({"Authorization": auth} if auth else {}), **({"Authorization": auth} if auth else {}),
} }
@@ -2930,13 +3111,34 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
direct_endpoint=target_url, direct_endpoint=target_url,
) )
started = time.monotonic() started = time.monotonic()
relay_ws_holder: list[Any] = []
frames = _relay_http_request_frames( frames = _relay_http_request_frames(
node.relay_addr, node.relay_addr,
path="/v1/chat/completions", path="/v1/chat/completions",
body=raw_body, body=raw_body,
headers=relay_headers, headers=relay_headers,
cancel_event=proxy_ctx.cancel_event,
ws_holder=relay_ws_holder,
ws_lock=proxy_ctx.relay_ws_lock,
) )
first = next(frames, None) first = next(frames, None)
with proxy_ctx.relay_ws_lock:
proxy_ctx.relay_ws = relay_ws_holder[0] if relay_ws_holder else None
if proxy_ctx.cancel_event.is_set():
if self._finalize_proxy_cancel(
proxy_ctx=proxy_ctx,
server=server,
request_id=request_id,
started=started,
model=model,
route_model=route_model,
route_nodes=route_nodes,
api_key=api_key,
node_work=node_work,
body=body,
finish_proxy_inflight=finish_proxy_inflight,
):
return
if first is not None and first.get("stream"): if first is not None and first.get("stream"):
# Streamed response (US-036): forward SSE chunks as they arrive # Streamed response (US-036): forward SSE chunks as they arrive
# and run the same token accounting as the direct stream path. # and run the same token accounting as the direct stream path.
@@ -2945,6 +3147,8 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
model, route_model, route_nodes, api_key, node_work, model, route_model, route_nodes, api_key, node_work,
request_body=body, request_body=body,
request_id=request_id, request_id=request_id,
proxy_ctx=proxy_ctx,
finish_proxy_inflight=finish_proxy_inflight,
) )
finish_proxy_inflight() finish_proxy_inflight()
return return
@@ -2959,6 +3163,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
in_tokens, out_tokens = 0, 0 in_tokens, out_tokens = 0, 0
tokens = in_tokens + out_tokens tokens = in_tokens + out_tokens
self._record_observed_throughput(model, route_model, tokens, elapsed, route_nodes) self._record_observed_throughput(model, route_model, tokens, elapsed, route_nodes)
_clear_proxy_progress_log_state(server, request_id)
_tracker_log( _tracker_log(
server, server,
"info", "info",
@@ -2989,7 +3194,73 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
try: try:
started = time.monotonic() started = time.monotonic()
upstream = urllib.request.urlopen(req, timeout=300.0) upstream_result: list[Any] = []
connect_errors: list[BaseException] = []
def _connect_upstream() -> None:
try:
upstream_result.append(urllib.request.urlopen(req, timeout=300.0))
except BaseException as exc:
connect_errors.append(exc)
connect_thread = threading.Thread(target=_connect_upstream, daemon=True)
connect_thread.start()
while connect_thread.is_alive():
if proxy_ctx.cancel_event.is_set():
connect_thread.join(timeout=310.0)
if upstream_result:
try:
upstream_result[0].close()
except Exception:
pass
if self._finalize_proxy_cancel(
proxy_ctx=proxy_ctx,
server=server,
request_id=request_id,
started=started,
model=model,
route_model=route_model,
route_nodes=route_nodes,
api_key=api_key,
node_work=node_work,
body=body,
finish_proxy_inflight=finish_proxy_inflight,
):
return
connect_thread.join(0.2)
if proxy_ctx.cancel_event.is_set():
if upstream_result:
try:
upstream_result[0].close()
except Exception:
pass
if self._finalize_proxy_cancel(
proxy_ctx=proxy_ctx,
server=server,
request_id=request_id,
started=started,
model=model,
route_model=route_model,
route_nodes=route_nodes,
api_key=api_key,
node_work=node_work,
body=body,
finish_proxy_inflight=finish_proxy_inflight,
):
return
if connect_errors:
raise connect_errors[0]
upstream = upstream_result[0]
with proxy_ctx.upstream_lock:
proxy_ctx.upstream = upstream
upstream_sock = _upstream_socket(upstream)
if upstream_sock is not None:
_set_upstream_read_timeout(upstream, None)
else:
_set_upstream_read_timeout(upstream, 0.5)
_tracker_log(server, "info", "proxy connected", request_id=request_id, target_url=target_url) _tracker_log(server, "info", "proxy connected", request_id=request_id, target_url=target_url)
except urllib.error.HTTPError as exc: except urllib.error.HTTPError as exc:
# Relay error status + body from node # Relay error status + body from node
@@ -3002,9 +3273,11 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
self.wfile.write(err_body) self.wfile.write(err_body)
except BrokenPipeError: except BrokenPipeError:
pass pass
_clear_proxy_progress_log_state(server, request_id)
finish_proxy_inflight() finish_proxy_inflight()
return return
except Exception as exc: except Exception as exc:
_clear_proxy_progress_log_state(server, request_id)
if node.relay_addr: if node.relay_addr:
_tracker_log( _tracker_log(
server, server,
@@ -3045,8 +3318,19 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
client_gone = False client_gone = False
try: try:
while True: while True:
line = upstream.readline() if proxy_ctx.cancel_event.is_set():
break
if upstream_sock is not None:
readable, _, _ = select.select([upstream_sock], [], [], 0.5)
if not readable:
continue
try:
line = upstream.readline()
except TimeoutError:
continue
if not line: if not line:
if proxy_ctx.cancel_event.is_set():
break
break break
if not client_gone: if not client_gone:
try: try:
@@ -3071,6 +3355,22 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
stream_usage = usage stream_usage = usage
except (BrokenPipeError, ConnectionResetError): except (BrokenPipeError, ConnectionResetError):
pass pass
if self._finalize_proxy_cancel(
proxy_ctx=proxy_ctx,
server=server,
request_id=request_id,
started=started,
model=model,
route_model=route_model,
route_nodes=route_nodes,
api_key=api_key,
node_work=node_work,
body=body,
finish_proxy_inflight=finish_proxy_inflight,
observed_stream_tokens=observed_stream_tokens,
stream_usage=stream_usage,
):
return
elapsed = time.monotonic() - started elapsed = time.monotonic() - started
# Bill even on client disconnect — the nodes did the work. # Bill even on client disconnect — the nodes did the work.
# Observed stream chunks are authoritative for the upper bound; # Observed stream chunks are authoritative for the upper bound;
@@ -3082,6 +3382,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
model, route_model, in_tokens + out_tokens, elapsed, route_nodes model, route_model, in_tokens + out_tokens, elapsed, route_nodes
) )
tokens = in_tokens + out_tokens tokens = in_tokens + out_tokens
_clear_proxy_progress_log_state(server, request_id)
_tracker_log( _tracker_log(
server, server,
"info", "info",
@@ -3113,6 +3414,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
in_tokens, out_tokens = 0, 0 in_tokens, out_tokens = 0, 0
tokens = in_tokens + out_tokens tokens = in_tokens + out_tokens
self._record_observed_throughput(model, route_model, tokens, elapsed, route_nodes) self._record_observed_throughput(model, route_model, tokens, elapsed, route_nodes)
_clear_proxy_progress_log_state(server, request_id)
_tracker_log( _tracker_log(
server, server,
"info", "info",
@@ -3272,6 +3574,9 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
node_work: list, node_work: list,
request_body: dict, request_body: dict,
request_id: str, request_id: str,
*,
proxy_ctx: _ActiveProxyContext | None = None,
finish_proxy_inflight: Any = None,
) -> None: ) -> None:
"""Forward a streamed relay response (US-036) to the client as SSE, """Forward a streamed relay response (US-036) to the client as SSE,
billing with the same accounting as the direct stream path.""" billing with the same accounting as the direct stream path."""
@@ -3285,6 +3590,8 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
observed_stream_tokens = 0 observed_stream_tokens = 0
client_gone = False client_gone = False
for frame in itertools.chain([first], frames): for frame in itertools.chain([first], frames):
if proxy_ctx is not None and proxy_ctx.cancel_event.is_set():
break
chunk = frame.get("chunk") or "" chunk = frame.get("chunk") or ""
if not chunk: if not chunk:
continue continue
@@ -3312,6 +3619,26 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
) )
if usage is not None: if usage is not None:
stream_usage = usage stream_usage = usage
if (
proxy_ctx is not None
and finish_proxy_inflight is not None
and self._finalize_proxy_cancel(
proxy_ctx=proxy_ctx,
server=server,
request_id=request_id,
started=started,
model=model,
route_model=route_model,
route_nodes=route_nodes,
api_key=api_key,
node_work=node_work,
body=request_body,
finish_proxy_inflight=finish_proxy_inflight,
observed_stream_tokens=observed_stream_tokens,
stream_usage=stream_usage,
)
):
return
elapsed = time.monotonic() - started elapsed = time.monotonic() - started
in_tokens, out_tokens = _stream_billable_split( in_tokens, out_tokens = _stream_billable_split(
observed_stream_tokens, stream_usage, request_body observed_stream_tokens, stream_usage, request_body
@@ -3320,6 +3647,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
model, route_model, in_tokens + out_tokens, elapsed, route_nodes model, route_model, in_tokens + out_tokens, elapsed, route_nodes
) )
tokens = in_tokens + out_tokens tokens = in_tokens + out_tokens
_clear_proxy_progress_log_state(server, request_id)
_tracker_log( _tracker_log(
server, server,
"info", "info",
@@ -3765,6 +4093,68 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
events = [dict(event) for event in server.console_events] events = [dict(event) for event in server.console_events]
self._send_json(200, {"events": events}) self._send_json(200, {"events": events})
def _handle_proxy_request_cancel(self, request_id: str) -> None:
server: _TrackerHTTPServer = self.server # type: ignore[assignment]
if server.accounts is not None and not self._require_role("admin"):
return
if not _request_proxy_cancel(server, request_id):
self._send_json(404, {"error": f"no active proxy for request {request_id!r}"})
return
self._send_json(200, {"status": "canceled", "request_id": request_id})
def _finalize_proxy_cancel(
self,
*,
proxy_ctx: _ActiveProxyContext,
server: "_TrackerHTTPServer",
request_id: str,
started: float,
model: str,
route_model: str,
route_nodes: list,
api_key: str | None,
node_work: list,
body: dict,
finish_proxy_inflight: Any,
observed_stream_tokens: int = 0,
stream_usage: dict | None = None,
) -> bool:
if not proxy_ctx.cancel_event.is_set():
return False
elapsed = time.monotonic() - started
_clear_proxy_progress_log_state(server, request_id)
tokens = observed_stream_tokens
if observed_stream_tokens > 0:
in_tokens, out_tokens = _stream_billable_split(
observed_stream_tokens, stream_usage, body,
)
tokens = in_tokens + out_tokens
self._record_observed_throughput(
model, route_model, tokens, elapsed, route_nodes,
)
_tracker_log(
server,
"info",
"proxy canceled",
request_id=request_id,
model=model,
route_model=route_model,
tokens=tokens,
elapsed_seconds=round(elapsed, 4),
tokens_per_sec=round(tokens / elapsed, 4) if elapsed > 0 and tokens > 0 else 0.0,
route=_node_route_summary(route_nodes),
)
if observed_stream_tokens > 0:
in_tokens, out_tokens = _stream_billable_split(
observed_stream_tokens, stream_usage, body,
)
self._bill_completed(
api_key, model, in_tokens + out_tokens, node_work,
input_tokens=in_tokens, output_tokens=out_tokens,
)
finish_proxy_inflight()
return True
def _handle_registry_wallets(self): def _handle_registry_wallets(self):
server: _TrackerHTTPServer = self.server # type: ignore[assignment] server: _TrackerHTTPServer = self.server # type: ignore[assignment]
if not self._require_role("admin"): if not self._require_role("admin"):
@@ -3855,7 +4245,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
server: _TrackerHTTPServer = self.server # type: ignore[assignment] server: _TrackerHTTPServer = self.server # type: ignore[assignment]
if server.accounts is None: if server.accounts is None:
return None return None
return server.accounts.session_account(_api_key_from_headers(self.headers)) return server.accounts.session_account(_session_token_from_headers(self.headers))
def _require_accounts(self) -> "AccountStore | None": def _require_accounts(self) -> "AccountStore | None":
server: _TrackerHTTPServer = self.server # type: ignore[assignment] server: _TrackerHTTPServer = self.server # type: ignore[assignment]
@@ -3895,7 +4285,11 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
f"[tracker] account registered: {account.get('email') or account.get('wallet')} " f"[tracker] account registered: {account.get('email') or account.get('wallet')} "
f"role={account['role']}", flush=True, f"role={account['role']}", flush=True,
) )
self._send_json(200, {"account": account, "session_token": token, "api_key": api_key}) self._send_json(
200,
{"account": account, "session_token": token, "api_key": api_key},
headers={"Set-Cookie": _session_cookie_header(token)},
)
def _handle_auth_login(self): def _handle_auth_login(self):
accounts = self._require_accounts() accounts = self._require_accounts()
@@ -3910,14 +4304,18 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
self._send_json(401, {"error": "invalid credentials"}) self._send_json(401, {"error": "invalid credentials"})
return return
token = accounts.create_session(account["account_id"]) token = accounts.create_session(account["account_id"])
self._send_json(200, {"account": account, "session_token": token}) self._send_json(
200,
{"account": account, "session_token": token},
headers={"Set-Cookie": _session_cookie_header(token)},
)
def _handle_auth_logout(self): def _handle_auth_logout(self):
accounts = self._require_accounts() accounts = self._require_accounts()
if accounts is None: if accounts is None:
return return
accounts.destroy_session(_api_key_from_headers(self.headers)) accounts.destroy_session(_session_token_from_headers(self.headers))
self._send_json(200, {"ok": True}) self._send_json(200, {"ok": True}, headers={"Set-Cookie": _session_cookie_header(None)})
def _handle_account_me(self): def _handle_account_me(self):
"""Balance, usage, and API keys for the logged-in account.""" """Balance, usage, and API keys for the logged-in account."""
@@ -4532,6 +4930,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
"model": resolved_name, "model": resolved_name,
"model_layers_end": required_end, "model_layers_end": required_end,
"peers": peers, "peers": peers,
"bytes_per_layer": _preset_bytes_per_layer(preset),
"model_sources": self._model_sources( "model_sources": self._model_sources(
resolved_name, resolved_name,
preset, preset,

View File

@@ -5,6 +5,7 @@ register/login/logout, per-account balance and usage, API-key lifecycle
(revoked keys rejected by the OpenAI proxy), and the admin listing. (revoked keys rejected by the OpenAI proxy), and the admin listing.
""" """
import http.cookies
import json import json
import urllib.error import urllib.error
import urllib.request import urllib.request
@@ -68,6 +69,17 @@ def test_sessions_resolve_and_destroy():
assert store.session_account("bogus") is None assert store.session_account("bogus") is None
def test_sessions_persist_across_restart(tmp_path):
db = str(tmp_path / "accounts.db")
store = AccountStore(db_path=db)
account = store.register(email="cookie@example.com", password="secret-123")
token = store.create_session(account["account_id"])
store.save_to_db()
reloaded = AccountStore(db_path=db)
assert reloaded.session_account(token)["account_id"] == account["account_id"]
def test_api_key_lifecycle(): def test_api_key_lifecycle():
store = AccountStore() store = AccountStore()
account = store.register(email="k@example.com", password="secret-123") account = store.register(email="k@example.com", password="secret-123")
@@ -156,6 +168,59 @@ def test_register_login_and_account_view(account_tracker):
assert me["usage"]["requests"] == 0 assert me["usage"]["requests"] == 0
def test_login_sets_cookie_and_cookie_auth_survives_tracker_restart(tmp_path):
accounts_db = str(tmp_path / "accounts.db")
tracker = TrackerServer(
billing=BillingLedger(starting_credit=0.0, default_price_per_1k=0.02),
accounts_db=accounts_db,
starting_credit=0.0,
devnet_topup_amount=0.0,
)
port = tracker.start()
url = f"http://127.0.0.1:{port}"
try:
_call(f"{url}/v1/auth/register", "POST",
{"email": "cookie-http@example.com", "password": "secret-123"})
req = urllib.request.Request(
f"{url}/v1/auth/login",
data=json.dumps({
"identifier": "cookie-http@example.com",
"password": "secret-123",
}).encode(),
headers={"Content-Type": "application/json"},
method="POST",
)
with urllib.request.urlopen(req) as r:
assert json.loads(r.read())["session_token"]
cookie_header = r.headers["Set-Cookie"]
finally:
tracker.stop()
cookie = http.cookies.SimpleCookie(cookie_header)
session_cookie = cookie["meshnet_session"].OutputString()
restarted = TrackerServer(
billing=BillingLedger(starting_credit=0.0, default_price_per_1k=0.02),
accounts_db=accounts_db,
starting_credit=0.0,
devnet_topup_amount=0.0,
)
restarted_port = restarted.start()
restarted_url = f"http://127.0.0.1:{restarted_port}"
try:
req = urllib.request.Request(
f"{restarted_url}/v1/account",
headers={"Cookie": session_cookie},
method="GET",
)
with urllib.request.urlopen(req) as r:
me = json.loads(r.read())
finally:
restarted.stop()
assert me["account"]["email"] == "cookie-http@example.com"
def test_bad_credentials_and_missing_session_are_401(account_tracker): def test_bad_credentials_and_missing_session_are_401(account_tracker):
url, _ = account_tracker url, _ = account_tracker
_call(f"{url}/v1/auth/register", "POST", _call(f"{url}/v1/auth/register", "POST",

View File

@@ -33,6 +33,21 @@ def test_dashboard_served_with_all_panels():
tracker.stop() tracker.stop()
def test_dashboard_chat_uses_streaming_fetch():
tracker = TrackerServer(billing=BillingLedger())
port = tracker.start()
try:
html = urllib.request.urlopen(
f"http://127.0.0.1:{port}/dashboard"
).read().decode()
finally:
tracker.stop()
assert "stream: true" in html
assert ".body.getReader()" in html
assert 'data === "[DONE]"' in html
def test_dashboard_served_by_follower(): def test_dashboard_served_by_follower():
"""A tracker that is not the leader (unreachable peers → never elected) """A tracker that is not the leader (unreachable peers → never elected)
still serves the dashboard from its own replicated state.""" still serves the dashboard from its own replicated state."""

View File

@@ -1118,7 +1118,7 @@ def test_real_model_startup_summary_shows_total_layers(tmp_path, monkeypatch, ca
assert captured_registration["vram_bytes"] == 6144 * 1024 * 1024 assert captured_registration["vram_bytes"] == 6144 * 1024 * 1024
assert captured_registration["max_loaded_shards"] == 2 assert captured_registration["max_loaded_shards"] == 2
output = capsys.readouterr().out output = capsys.readouterr().out
assert "Shard: layers 023; 24 of 24" in output assert "Shard: layers 023 (24 of 24)" in output
assert "Node ID: node-test-123" in output assert "Node ID: node-test-123" in output
@@ -1646,6 +1646,106 @@ def test_preset_model_startup_honors_pinned_shard_range(tmp_path, monkeypatch):
tracker.stop() tracker.stop()
def test_preset_startup_rejects_pinned_shard_above_memory_budget(tmp_path, monkeypatch):
"""Pinned layer ranges that exceed the node memory budget fail before model load."""
import meshnet_node.startup as startup_mod
monkeypatch.setattr(
startup_mod,
"detect_hardware",
lambda: {"device": "cpu", "gpu_name": None, "vram_mb": 0, "ram_mb": 8 * 1024},
)
tracker = TrackerServer(model_presets={
"big-model": {
"layers_start": 0,
"layers_end": 39,
"hf_repo": "org/big-model",
"bytes_per_layer": {"bfloat16": 2 * 1024 * 1024 * 1024},
},
})
tracker_port = tracker.start()
tracker_url = f"http://127.0.0.1:{tracker_port}"
try:
with pytest.raises(ValueError, match="Pinned shard layers 039"):
run_startup(
tracker_url=tracker_url,
model="big-model",
shard_start=0,
shard_end=39,
wallet_path=tmp_path / "wallet.json",
cache_dir=tmp_path / "shards",
)
finally:
tracker.stop()
def test_preset_model_with_hf_repo_loads_torch_backend(tmp_path, monkeypatch, capsys):
"""Named presets that advertise hf_repo must load TorchNodeServer, not the stub server."""
import meshnet_node.startup as startup_mod
class FakeBackend:
total_layers = 16
torch_calls: list[dict] = []
class FakeTorchNodeServer:
def __init__(self, **kwargs):
torch_calls.append(kwargs)
self.backend = FakeBackend()
self.port = None
self.chat_completion_count = 0
self.tracker_node_id = None
def start(self):
self.port = 7002
return self.port
def stop(self):
pass
monkeypatch.setattr(
startup_mod,
"detect_hardware",
lambda: {"device": "cpu", "gpu_name": None, "vram_mb": 0, "ram_mb": 16 * 1024},
)
monkeypatch.setattr(startup_mod, "TorchNodeServer", FakeTorchNodeServer)
monkeypatch.setattr(startup_mod, "StubNodeServer", lambda **_kw: (_ for _ in ()).throw(AssertionError("preset with hf_repo must not use StubNodeServer")))
model_dir = tmp_path / "node-shards" / "tiny-llama"
model_dir.mkdir(parents=True)
(model_dir / "config.json").write_text('{"num_hidden_layers": 16}')
monkeypatch.setattr(startup_mod, "download_shard", lambda *_a, **_kw: model_dir)
tracker = TrackerServer(model_presets={
"tiny-llama": {"layers_start": 0, "layers_end": 15, "hf_repo": "org/tiny-llama-shards"}
})
tracker_port = tracker.start()
tracker_url = f"http://127.0.0.1:{tracker_port}"
try:
node = run_startup(
tracker_url=tracker_url,
model="tiny-llama",
wallet_path=tmp_path / "wallet.json",
cache_dir=tmp_path / "node-shards",
)
try:
assert len(torch_calls) == 1
assert torch_calls[0]["model_id"] == "org/tiny-llama-shards"
assert torch_calls[0]["cache_dir"] == model_dir
output = capsys.readouterr().out
assert "Loading real PyTorch model shard..." in output
assert "Model ID: org/tiny-llama-shards" in output
network_map = _get_json(f"{tracker_url}/v1/network/map")
registered = network_map["nodes"][0]
assert registered["hf_repo"] == "org/tiny-llama-shards"
assert registered["num_layers"] == 16
finally:
node.stop()
finally:
tracker.stop()
def test_torch_startup_retries_registration_when_tracker_unreachable( def test_torch_startup_retries_registration_when_tracker_unreachable(
tmp_path, tmp_path,
monkeypatch, monkeypatch,

File diff suppressed because it is too large Load Diff

View File

@@ -142,6 +142,21 @@ def _send_chat_request(gateway_url: str, prompt: str) -> dict:
return json.loads(r.read()) return json.loads(r.read())
def _send_streaming_chat_request(gateway_url: str, prompt: str):
data = json.dumps({
"model": GPT2_MODEL,
"messages": [{"role": "user", "content": prompt}],
"stream": True,
}).encode()
req = urllib.request.Request(
f"{gateway_url}/v1/chat/completions",
data=data,
headers={"Content-Type": "application/json"},
method="POST",
)
return urllib.request.urlopen(req)
def test_all_responses_valid_openai_format(tracker_node_setup): def test_all_responses_valid_openai_format(tracker_node_setup):
"""Ten requests via gateway all return valid OpenAI chat completion format.""" """Ten requests via gateway all return valid OpenAI chat completion format."""
gateway_url, _, _ = tracker_node_setup gateway_url, _, _ = tracker_node_setup
@@ -155,6 +170,30 @@ def test_all_responses_valid_openai_format(tracker_node_setup):
assert isinstance(message.get("content"), str), f"request {i}: content must be a string" assert isinstance(message.get("content"), str), f"request {i}: content must be a string"
def test_streaming_head_worker_response_is_not_buffered_with_content_length(tracker_node_setup):
"""Gateway must relay head-worker SSE as a live stream, not a buffered JSON-sized body."""
gateway_url, _, _ = tracker_node_setup
with _send_streaming_chat_request(gateway_url, "stream through head worker") as resp:
assert resp.status == 200
assert "text/event-stream" in resp.headers["Content-Type"]
assert "Content-Length" not in resp.headers
data_lines = []
while len(data_lines) < 4:
line = resp.readline().decode().strip()
if line.startswith("data: "):
data_lines.append(line)
if line == "data: [DONE]":
break
assert data_lines[-1] == "data: [DONE]"
content = "".join(
json.loads(line[6:])["choices"][0].get("delta", {}).get("content", "")
for line in data_lines[:-1]
)
assert "head-worker" in content
def test_both_tracker_nodes_receive_load(tracker_node_setup): def test_both_tracker_nodes_receive_load(tracker_node_setup):
"""Both head workers handle at least one request each out of ten.""" """Both head workers handle at least one request each out of ten."""
gateway_url, tracker_node_a, tracker_node_b = tracker_node_setup gateway_url, tracker_node_a, tracker_node_b = tracker_node_setup

View File

@@ -0,0 +1,67 @@
import logging
import sys
from meshnet_tracker.logging_setup import configure_tracker_file_logging, tracker_logger
def test_tracker_file_logging_writes_separate_level_files(tmp_path):
original_stdout = sys.stdout
original_stderr = sys.stderr
try:
log_dir = configure_tracker_file_logging(tmp_path, tee_stdio=False)
logger = tracker_logger()
logger.info("info-event")
logger.warning("warning-event")
logger.error("error-event")
for handler in logger.handlers:
handler.flush()
assert (log_dir / "info.log").read_text().count("info-event") == 1
assert "warning-event" not in (log_dir / "info.log").read_text()
assert "error-event" not in (log_dir / "info.log").read_text()
assert "warning-event" in (log_dir / "warning.log").read_text()
assert "info-event" not in (log_dir / "warning.log").read_text()
assert "error-event" not in (log_dir / "warning.log").read_text()
assert "error-event" in (log_dir / "error.log").read_text()
assert "info-event" not in (log_dir / "error.log").read_text()
assert "warning-event" not in (log_dir / "error.log").read_text()
finally:
sys.stdout = original_stdout
sys.stderr = original_stderr
def test_tracker_file_logging_tees_stdio_and_rotates(tmp_path):
original_stdout = sys.stdout
original_stderr = sys.stderr
try:
log_dir = configure_tracker_file_logging(
tmp_path,
max_bytes=120,
backup_count=1,
)
print("stdout goes to info", flush=True)
print("stderr goes to error", file=sys.stderr, flush=True)
for handler in tracker_logger().handlers:
handler.flush()
assert "stdout goes to info" in (log_dir / "info.log").read_text()
assert "stderr goes to error" in (log_dir / "error.log").read_text()
for index in range(12):
tracker_logger().info("rotating-info-line-%02d", index)
for handler in tracker_logger().handlers:
handler.flush()
assert (log_dir / "info.log.1").exists()
finally:
sys.stdout = original_stdout
sys.stderr = original_stderr
logger = tracker_logger()
for handler in logger.handlers:
handler.close()
logger.handlers.clear()
logger.setLevel(logging.NOTSET)

View File

@@ -5,6 +5,7 @@ import json
import threading import threading
import time import time
import urllib.error import urllib.error
import urllib.parse
import urllib.request import urllib.request
import pytest import pytest
@@ -501,6 +502,169 @@ def test_tracker_logs_stream_progress_before_request_completes():
node_thread.join(timeout=1.0) node_thread.join(timeout=1.0)
def test_tracker_stream_survives_idle_gap_between_sse_chunks():
first_chunk_sent = threading.Event()
class IdleStreamingChatHandler(http.server.BaseHTTPRequestHandler):
def log_message(self, format, *args):
pass
def do_POST(self):
if self.path != "/v1/chat/completions":
self.send_response(404)
self.end_headers()
return
self.rfile.read(int(self.headers.get("Content-Length", 0)))
self.send_response(200)
self.send_header("Content-Type", "text/event-stream; charset=utf-8")
self.end_headers()
first = json.dumps({
"choices": [{"delta": {"content": "hello"}}],
}).encode()
second = json.dumps({
"choices": [{"delta": {"content": " world"}}],
}).encode()
self.wfile.write(b"data: " + first + b"\n\n")
self.wfile.flush()
first_chunk_sent.set()
time.sleep(1.0)
self.wfile.write(b"data: " + second + b"\n\n")
self.wfile.write(b"data: [DONE]\n\n")
self.wfile.flush()
node = http.server.HTTPServer(("127.0.0.1", 0), IdleStreamingChatHandler)
node_thread = threading.Thread(target=node.serve_forever, daemon=True)
node_thread.start()
tracker = TrackerServer(heartbeat_timeout=60.0)
tracker_port = tracker.start()
response = None
try:
_post_json(
f"http://127.0.0.1:{tracker_port}/v1/nodes/register",
{"endpoint": f"http://127.0.0.1:{node.server_address[1]}",
"model": "idle-stream-model", "num_layers": 1,
"shard_start": 0, "shard_end": 0,
"hardware_profile": {}, "score": 1.0},
)
req = urllib.request.Request(
f"http://127.0.0.1:{tracker_port}/v1/chat/completions",
data=json.dumps({
"model": "idle-stream-model",
"stream": True,
"messages": [{"role": "user", "content": "hi"}],
}).encode(),
headers={"Content-Type": "application/json"},
method="POST",
)
response = urllib.request.urlopen(req, timeout=3.0)
assert response.readline().startswith(b"data:")
assert first_chunk_sent.wait(timeout=1.0)
remaining = response.read().splitlines()
assert b"data: [DONE]" in remaining
finally:
if response is not None:
response.close()
tracker.stop()
node.shutdown()
node.server_close()
node_thread.join(timeout=1.0)
def test_tracker_dashboard_can_cancel_inflight_proxy():
chunk_sent = threading.Event()
release = threading.Event()
class StreamingChatHandler(http.server.BaseHTTPRequestHandler):
def log_message(self, fmt, *args):
pass
def do_POST(self):
if self.path != "/v1/chat/completions":
self.send_response(404)
self.end_headers()
return
self.rfile.read(int(self.headers.get("Content-Length", 0)))
self.send_response(200)
self.send_header("Content-Type", "text/event-stream; charset=utf-8")
self.end_headers()
payload = json.dumps({
"choices": [{"delta": {"content": "hello world"}}],
}).encode()
self.wfile.write(b"data: " + payload + b"\n\n")
self.wfile.flush()
chunk_sent.set()
release.wait(timeout=3.0)
self.wfile.write(b"data: [DONE]\n\n")
self.wfile.flush()
node = http.server.HTTPServer(("127.0.0.1", 0), StreamingChatHandler)
node_thread = threading.Thread(target=node.serve_forever, daemon=True)
node_thread.start()
tracker = TrackerServer(heartbeat_timeout=60.0)
tracker_port = tracker.start()
response = None
request_id = None
try:
_post_json(
f"http://127.0.0.1:{tracker_port}/v1/nodes/register",
{"endpoint": f"http://127.0.0.1:{node.server_address[1]}",
"model": "cancel-proxy-model", "num_layers": 1,
"shard_start": 0, "shard_end": 0,
"hardware_profile": {}, "score": 1.0},
)
req = urllib.request.Request(
f"http://127.0.0.1:{tracker_port}/v1/chat/completions",
data=json.dumps({
"model": "cancel-proxy-model",
"stream": True,
"messages": [{"role": "user", "content": "hi"}],
}).encode(),
headers={"Content-Type": "application/json"},
method="POST",
)
response = urllib.request.urlopen(req, timeout=3.0)
first_line = response.readline()
assert first_line.startswith(b"data:")
assert chunk_sent.wait(timeout=1.0)
console = _get_json(f"http://127.0.0.1:{tracker_port}/v1/console")
selected = [
event for event in console["events"]
if event["message"] == "proxy route selected"
]
assert selected
request_id = selected[-1]["fields"]["request_id"]
cancel = _post_json(
f"http://127.0.0.1:{tracker_port}/v1/proxy/requests/{urllib.parse.quote(request_id, safe='')}/cancel",
{},
)
assert cancel["status"] == "canceled"
deadline = time.time() + 5.0
canceled_events = []
while time.time() < deadline:
console = _get_json(f"http://127.0.0.1:{tracker_port}/v1/console")
canceled_events = [
event for event in console["events"]
if event["message"] == "proxy canceled"
and event["fields"].get("request_id") == request_id
]
if canceled_events:
break
time.sleep(0.05)
assert canceled_events
finally:
release.set()
if response is not None:
response.close()
tracker.stop()
node.shutdown()
node.server_close()
node_thread.join(timeout=1.0)
def test_tracker_routes_hf_model_alias_from_quickstart(): def test_tracker_routes_hf_model_alias_from_quickstart():
"""The documented qwen2.5-0.5b alias resolves a full HF repo registration.""" """The documented qwen2.5-0.5b alias resolves a full HF repo registration."""
tracker = TrackerServer() tracker = TrackerServer()