popov/neuron-tai
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: popov:3eb7c6b93e8deee6ddde0d4de41f4a27a78cb05a
Branches Tags
popov:master popov:ralph/dgr-001-performance-contract popov:cursor/fix-meshnet-node-param-parsing popov:worktree-feat+us-016
...
compare: popov:fdeb881c8344010838db4060d84cd4ce7428a90d
Branches Tags
popov:ralph/dgr-001-performance-contract popov:master popov:cursor/fix-meshnet-node-param-parsing popov:worktree-feat+us-016

3 Commits
3eb7c6b93e ... fdeb881c83

Author SHA1 Message Date
Dobromir Popov
fdeb881c83 web UI 2026-07-07 17:54:22 +03:00
Dobromir Popov
08e9c22ccf Merge origin/master: streaming progress, dashboard call wall, and heartbeat scaffolding. 
Resolve conflicts in dashboard.html (Call wall + live TPS/queue from remote) and server.py (proxy progress logging, request id forwarding, current_requests on node entries).

Co-authored-by: Cursor <cursoragent@cursor.com>
 2026-07-07 17:44:18 +03:00
Dobromir Popov
e81d989f39 dash QOL 2026-07-07 17:37:38 +03:00
12 changed files with 1599 additions and 476 deletions
Expand all files Collapse all files

BIN
billing.sqlite
View File

Binary file not shown.

12
docs/issues/44-tracker-shard-source-partial-download.md
View File

@@ -75,7 +75,7 @@ What exists already (build on it, don't duplicate):
- [x] Node downloader keeps exact-shard peers first, then races tracker model - [x] Node downloader keeps exact-shard peers first, then races tracker model
sources against a HuggingFace `snapshot_download(..., allow_patterns=...)` sources against a HuggingFace `snapshot_download(..., allow_patterns=...)`
subset download, using the first successful source. subset download, using the first successful source.
- [ ] When no tracker model source is available at all, the HuggingFace - [x] When no tracker model source is available at all, the HuggingFace
fallback still computes `allow_patterns` from the repo's own fallback still computes `allow_patterns` from the repo's own
`model.safetensors.index.json` (fetched directly, not via the tracker) — `model.safetensors.index.json` (fetched directly, not via the tracker) —
it never silently downloads the full model just because the tracker has it never silently downloads the full model just because the tracker has
@@ -95,7 +95,9 @@ What exists already (build on it, don't duplicate):
- 2026-07-06: Added the tracker/node download path. For immediate Qwen3.6-35B - 2026-07-06: Added the tracker/node download path. For immediate Qwen3.6-35B
LAN testing, real PyTorch nodes fetch the full snapshot from the tracker via LAN testing, real PyTorch nodes fetch the full snapshot from the tracker via
`full_url` and race HuggingFace as fallback. Remaining hard half is true `full_url`; HuggingFace remains fallback-only, and when it is used the node
partial model materialization: the backend can prefer a downloaded local computes `allow_patterns` from the repo's remote SafeTensors index so it
model directory, but Transformers still needs a `meta`-device load path that stays layer-filtered even without tracker-cached files. Remaining hard half
materializes only assigned layers. is true partial model materialization: the backend can prefer a downloaded
local model directory, but Transformers still needs a `meta`-device load
path that materializes only assigned layers.

BIN
meshnet_registry.sqlite3
View File

Binary file not shown.

2
packages/node/meshnet_node/downloader.py
View File

@@ -1,4 +1,4 @@
"""Shard downloader — fetches model shards from peers or HuggingFace Hub. """Shard downloader — fetches model files from peers, tracker sources, or HuggingFace.
Cache layout: ~/.cache/meshnet/shards/<model>/ Cache layout: ~/.cache/meshnet/shards/<model>/

216
packages/node/meshnet_node/model_backend.py
View File

@@ -4,6 +4,7 @@ from __future__ import annotations
import base64 import base64
from dataclasses import dataclass from dataclasses import dataclass
import json
from pathlib import Path from pathlib import Path
from typing import Any, Literal from typing import Any, Literal
@@ -22,6 +23,10 @@ class InsufficientVRAMError(ModelBackendError):
"""Raised when a requested shard cannot fit in available CUDA memory.""" """Raised when a requested shard cannot fit in available CUDA memory."""
class PartialModelLoadUnsupported(ModelBackendError):
"""Raised when a shard cannot be materialized from a local snapshot subset."""
@dataclass(frozen=True) @dataclass(frozen=True)
class TensorPayload: class TensorPayload:
body: bytes body: bytes
@@ -94,20 +99,39 @@ class TorchModelShard:
None if load_source != model_id else cache_dir, None if load_source != model_id else cache_dir,
) )
try: try:
load_kwargs = { total_layers_hint = _total_layers_for_local_snapshot(AutoConfig, load_source)
"device_map": "auto" if uses_quantized_weights else None, if _should_partial_materialize_shard(
"dtype": dtype,
"low_cpu_mem_usage": True,
"cache_dir": str(cache_dir) if cache_dir is not None and load_source == model_id else None,
}
if quant_config is not None:
load_kwargs["quantization_config"] = quant_config
self.model = AutoModelForCausalLM.from_pretrained(
load_source, load_source,
**load_kwargs, shard_start,
) shard_end,
if not uses_quantized_weights: total_layers_hint=total_layers_hint,
self.model.to(self.device) uses_quantized_weights=uses_quantized_weights,
):
self.model = _load_partial_model_from_snapshot(
AutoConfig,
AutoModelForCausalLM,
torch,
load_source,
shard_start,
shard_end,
dtype,
self.device,
)
else:
load_kwargs = {
"device_map": "auto" if uses_quantized_weights else None,
"dtype": dtype,
"low_cpu_mem_usage": True,
"cache_dir": str(cache_dir) if cache_dir is not None and load_source == model_id else None,
}
if quant_config is not None:
load_kwargs["quantization_config"] = quant_config
self.model = AutoModelForCausalLM.from_pretrained(
load_source,
**load_kwargs,
)
if not uses_quantized_weights:
self.model.to(self.device)
except Exception as exc: except Exception as exc:
if _looks_like_oom(exc): if _looks_like_oom(exc):
raise InsufficientVRAMError( raise InsufficientVRAMError(
@@ -357,6 +381,135 @@ def load_torch_shard(
return TorchModelShard(model_id, shard_start, shard_end, quantization, cache_dir) return TorchModelShard(model_id, shard_start, shard_end, quantization, cache_dir)
def _total_layers_for_local_snapshot(auto_config: Any, load_source: str) -> int | None:
snapshot_dir = Path(load_source)
if not (snapshot_dir / "config.json").exists():
return None
from .model_catalog import layers_from_config
try:
cfg = auto_config.from_pretrained(str(snapshot_dir))
except Exception:
return None
return layers_from_config(cfg)
def _should_partial_materialize_shard(
load_source: str,
shard_start: int,
shard_end: int,
*,
total_layers_hint: int | None,
uses_quantized_weights: bool,
) -> bool:
if uses_quantized_weights:
return False
snapshot_dir = Path(load_source)
if not snapshot_dir.exists() or not (snapshot_dir / "config.json").exists():
return False
if not (snapshot_dir / "model.safetensors.index.json").exists():
return False
if total_layers_hint is None:
return False
return not (shard_start == 0 and shard_end >= total_layers_hint - 1)
def _load_partial_model_from_snapshot(
auto_config: Any,
auto_model_for_causal_lm: Any,
torch: Any,
load_source: str,
shard_start: int,
shard_end: int,
dtype: Any,
device: Any,
*,
init_empty_weights_fn: Any | None = None,
set_tensor_fn: Any | None = None,
safe_open_fn: Any | None = None,
) -> Any:
from .model_catalog import layers_from_config
from .safetensors_selection import (
INDEX_FILENAME,
select_tensor_names_for_layers_from_index,
)
if init_empty_weights_fn is None:
from accelerate import init_empty_weights as init_empty_weights_fn
if set_tensor_fn is None:
from accelerate.utils import set_module_tensor_to_device as set_tensor_fn
if safe_open_fn is None:
from safetensors import safe_open as safe_open_fn
snapshot_dir = Path(load_source)
cfg = auto_config.from_pretrained(str(snapshot_dir))
total_layers = layers_from_config(cfg)
if total_layers is None:
raise PartialModelLoadUnsupported(
f"could not determine num_hidden_layers for local snapshot {snapshot_dir}"
)
if shard_end >= total_layers:
raise ValueError(
f"shard_end {shard_end} exceeds last layer index {total_layers - 1}"
)
index_path = snapshot_dir / INDEX_FILENAME
try:
index = json.loads(index_path.read_text(encoding="utf-8"))
except FileNotFoundError as exc:
raise PartialModelLoadUnsupported(
f"missing SafeTensors index for partial load: {index_path}"
) from exc
weight_map = index.get("weight_map")
if not isinstance(weight_map, dict):
raise PartialModelLoadUnsupported(f"{INDEX_FILENAME} must contain a weight_map object")
tensor_names = select_tensor_names_for_layers_from_index(
weight_map,
shard_start,
shard_end,
total_layers=total_layers,
)
if not tensor_names:
raise PartialModelLoadUnsupported(
f"no checkpoint tensors matched layers {shard_start}-{shard_end} in {snapshot_dir}"
)
with init_empty_weights_fn():
model = auto_model_for_causal_lm.from_config(cfg, torch_dtype=dtype)
tie_weights = getattr(model, "tie_weights", None)
if callable(tie_weights):
tie_weights()
tensors_by_file: dict[str, list[str]] = {}
for tensor_name in sorted(tensor_names):
rel_file = weight_map.get(tensor_name)
if not isinstance(rel_file, str):
continue
tensors_by_file.setdefault(rel_file, []).append(tensor_name)
for rel_file, names in tensors_by_file.items():
checkpoint_file = snapshot_dir / rel_file
if not checkpoint_file.exists():
raise PartialModelLoadUnsupported(
f"checkpoint file advertised in {INDEX_FILENAME} is missing: {checkpoint_file}"
)
with safe_open_fn(str(checkpoint_file), framework="pt", device="cpu") as handle:
for tensor_name in names:
set_tensor_fn(
model,
tensor_name,
device,
value=handle.get_tensor(tensor_name),
dtype=dtype,
)
for module in _active_modules_for_shard(model, shard_start, shard_end):
if hasattr(module, "to"):
module.to(device)
return model
def _model_load_plan( def _model_load_plan(
auto_config: Any, auto_config: Any,
model_id: str, model_id: str,
@@ -442,6 +595,37 @@ def _position_embeddings(model: Any) -> Any | None:
return None return None
def _rotary_embedding_module(model: Any) -> Any | None:
if hasattr(model, "model") and hasattr(model.model, "rotary_emb"):
return model.model.rotary_emb
if hasattr(model, "transformer") and hasattr(model.transformer, "rotary_emb"):
return model.transformer.rotary_emb
return None
def _active_modules_for_shard(model: Any, shard_start: int, shard_end: int) -> list[Any]:
active: list[Any] = []
def add(module: Any | None) -> None:
if module is None:
return
if any(existing is module for existing in active):
return
active.append(module)
if shard_start == 0:
add(_embed_tokens(model))
add(_position_embeddings(model))
add(_rotary_embedding_module(model))
for layer in _model_layers(model)[shard_start:shard_end + 1]:
add(layer)
total_layers = len(_model_layers(model))
if shard_end >= total_layers - 1:
add(_final_norm(model))
add(getattr(model, "lm_head", None))
return active
def _final_norm(model: Any) -> Any | None: def _final_norm(model: Any) -> Any | None:
if hasattr(model, "model") and hasattr(model.model, "norm"): if hasattr(model, "model") and hasattr(model.model, "norm"):
return model.model.norm return model.model.norm
@@ -485,11 +669,7 @@ def _rotary_position_embeddings(model: Any, hidden_states: Any, position_ids: An
"""Return model-level rotary embeddings required by newer HF decoder layers.""" """Return model-level rotary embeddings required by newer HF decoder layers."""
if position_ids is None: if position_ids is None:
return None return None
rotary = None rotary = _rotary_embedding_module(model)
if hasattr(model, "model") and hasattr(model.model, "rotary_emb"):
rotary = model.model.rotary_emb
elif hasattr(model, "transformer") and hasattr(model.transformer, "rotary_emb"):
rotary = model.transformer.rotary_emb
if rotary is None: if rotary is None:
return None return None
return rotary(hidden_states, position_ids) return rotary(hidden_states, position_ids)

17
packages/node/meshnet_node/safetensors_selection.py
View File

@@ -118,6 +118,23 @@ def select_files_for_layers_from_index(
return selected return selected
def select_tensor_names_for_layers_from_index(
weight_map: dict[str, str],
start_layer: int,
end_layer: int,
*,
total_layers: int | None = None,
) -> set[str]:
"""Pure variant that returns checkpoint tensor names instead of file paths."""
selected: set[str] = set()
for tensor_name, rel_file in weight_map.items():
if not isinstance(tensor_name, str) or not isinstance(rel_file, str):
continue
if _tensor_belongs_to_range(tensor_name, start_layer, end_layer, total_layers):
selected.add(tensor_name)
return selected
def _tensor_belongs_to_range( def _tensor_belongs_to_range(
tensor_name: str, tensor_name: str,
start_layer: int, start_layer: int,

10
packages/tracker/meshnet_tracker/billing.py
View File

@@ -453,13 +453,13 @@ class BillingLedger:
with self._lock: with self._lock:
return self._node_pending.get(wallet, 0.0) return self._node_pending.get(wallet, 0.0)
def usage_for(self, api_keys: list[str], *, recent_limit: int = 20) -> dict: def usage_for(self, api_keys: list[str], *, recent_limit: int | None = None) -> dict:
"""Aggregate charge history for a set of API keys (dashboard view).""" """Aggregate charge history for a set of API keys (dashboard view)."""
keys = set(api_keys) keys = set(api_keys)
requests = 0 requests = 0
total_tokens = 0 total_tokens = 0
total_cost = 0.0 total_cost = 0.0
recent: list[dict] = [] records: list[dict] = []
with self._lock: with self._lock:
for event in self._event_log: for event in self._event_log:
if event.get("type") != "charge" or event.get("api_key") not in keys: if event.get("type") != "charge" or event.get("api_key") not in keys:
@@ -467,18 +467,20 @@ class BillingLedger:
requests += 1 requests += 1
total_tokens += int(event.get("total_tokens", 0)) total_tokens += int(event.get("total_tokens", 0))
total_cost += float(event.get("cost", 0.0)) total_cost += float(event.get("cost", 0.0))
recent.append({ records.append({
"api_key": event["api_key"], "api_key": event["api_key"],
"model": event.get("model"), "model": event.get("model"),
"total_tokens": event.get("total_tokens", 0), "total_tokens": event.get("total_tokens", 0),
"cost": event.get("cost", 0.0), "cost": event.get("cost", 0.0),
"ts": event.get("ts", 0.0), "ts": event.get("ts", 0.0),
}) })
recent = records[-recent_limit:] if recent_limit is not None else records
return { return {
"requests": requests, "requests": requests,
"total_tokens": total_tokens, "total_tokens": total_tokens,
"total_cost": total_cost, "total_cost": total_cost,
"recent": recent[-recent_limit:], "records": records,
"recent": recent,
} }
def snapshot(self) -> dict: def snapshot(self) -> dict:

711
packages/tracker/meshnet_tracker/dashboard.html
View File

@@ -39,12 +39,41 @@
.form-row { display:flex; gap:8px; } .form-row { display:flex; gap:8px; }
.form-row button { white-space:nowrap; } .form-row button { white-space:nowrap; }
.error-msg { color:var(--bad); font-size:12px; min-height:16px; } .error-msg { color:var(--bad); font-size:12px; min-height:16px; }
.keybox { word-break:break-all; background:var(--bg); border:1px solid var(--border); .keybox { display:flex; flex-wrap:wrap; align-items:center; gap:6px;
position:relative;
word-break:break-all; background:var(--bg); border:1px solid var(--border);
border-radius:6px; padding:4px 8px; margin:4px 0; font-size:11px; } border-radius:6px; padding:4px 8px; margin:4px 0; font-size:11px; }
.key-text { cursor:text; flex:1 1 auto; min-width:12rem; }
.copy-tooltip {
position:absolute; right:8px; top:-26px;
background:var(--panel); border:1px solid var(--border); color:var(--ok);
padding:2px 8px; border-radius:4px; font-size:11px;
pointer-events:none; z-index:1; white-space:nowrap;
}
.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 button { border:0; border-bottom:1px solid transparent; border-radius:0;
background:transparent; color:var(--dim); padding:5px 0 8px; }
.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; }
.chat-shell { display:grid; grid-template-columns:minmax(0, 1.35fr) minmax(320px, 0.65fr); gap:12px; }
.chat-pane { display:flex; flex-direction:column; gap:10px; min-width: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; }
.chat-controls label { display:flex; flex-direction:column; gap:4px; color:var(--dim); }
.chat-controls select { min-width:220px; }
.chat-history { display:flex; flex-direction:column; gap:8px; min-height:220px; max-height:420px; overflow:auto; }
.chat-message { border:1px solid #21262d; border-radius:6px; padding:8px 10px; background:#10151d; }
.chat-role { color:var(--dim); font-size:11px; text-transform:uppercase; letter-spacing:.06em; margin-bottom:4px; }
.chat-role-user { color:var(--accent); }
.chat-role-assistant { color:var(--ok); }
.chat-role-error { color:var(--bad); }
.chat-compose { display:flex; flex-direction:column; gap:8px; }
.chat-compose textarea { min-height:112px; resize:vertical; width:100%; }
.chat-status { color:var(--dim); font-size:12px; }
.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;
@@ -55,6 +84,10 @@
.console-level-info { color:var(--accent); } .console-level-info { color:var(--accent); }
.console-level-warn { color:var(--warn); } .console-level-warn { color:var(--warn); }
.console-level-error { color:var(--bad); } .console-level-error { color:var(--bad); }
.status-pending { color:var(--warn); }
.status-processing { color:var(--accent); }
.status-failed { color:var(--bad); }
.status-complete { color:var(--ok); }
</style> </style>
</head> </head>
<body> <body>
@@ -63,19 +96,52 @@
<span class="meta" id="self-url"></span> <span class="meta" id="self-url"></span>
<span class="meta" id="refreshed"></span> <span class="meta" id="refreshed"></span>
</header> </header>
<nav class="dashboard-tabs" aria-label="Dashboard sections">
<button id="tab-overview" class="active" onclick="switchDashboardTab('overview')">Overview</button>
<button id="tab-chat" onclick="switchDashboardTab('chat')">Chat</button>
<button id="tab-billing" style="display:none" onclick="switchDashboardTab('billing')">Billing</button>
<button id="tab-admin" style="display:none" onclick="switchDashboardTab('admin')">Admin</button>
</nav>
<main> <main>
<section id="account-section"><h2>Account</h2><div id="account">loading…</div></section> <section data-tab="overview" id="account-section"><h2>Account</h2><div id="account">loading…</div></section>
<section id="admin-section" style="display:none"><h2>All accounts (admin)</h2><div id="admin" class="empty"></div></section> <section data-tab="overview"><h2>Tracker hive</h2><div id="hive" class="empty">loading…</div></section>
<section><h2>Tracker hive</h2><div id="hive" class="empty">loading…</div></section> <section data-tab="overview"><h2>Nodes &amp; coverage</h2><div id="nodes" class="empty">loading…</div></section>
<section><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><h2>Client balances</h2><div id="clients" 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><h2>Node pending payouts</h2><div id="pending" class="empty">loading…</div></section> <section data-tab="chat" class="wide">
<section><h2>Settlement history</h2><div id="settlements" class="empty">loading…</div></section> <h2>Chat / inference</h2>
<section><h2>Strikes / bans / forfeitures</h2><div id="fraud" class="empty">loading…</div></section> <div class="chat-shell">
<section><h2>Model usage (RPM)</h2><div id="stats" class="empty">loading…</div></section> <div class="chat-pane">
<section><h2>Node throughput</h2><div id="throughput" class="empty">loading…</div></section> <div class="chat-panel chat-controls">
<section class="wide"><h2>Console output</h2><div id="console" class="console empty">loading…</div></section> <label>Model
<section class="wide"><h2>Inference history</h2><div id="inference-history" class="empty">loading...</div></section> <select id="chat-model" onchange="selectChatModel(this.value)"></select>
</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-history" class="chat-history empty">no messages yet</div>
</div>
</div>
</div>
</section>
<section data-tab="billing" data-logged-in-only><h2>Usage summary</h2><div id="usage-summary" class="empty">login required</div></section>
<section data-tab="billing" data-logged-in-only><h2>Node throughput</h2><div id="node-throughput" class="empty">login required</div></section>
<section data-tab="billing"><h2>Request history</h2><div id="billing-usage" class="empty">login required</div></section>
<section data-tab="billing" data-admin-only><h2>Node pending payouts</h2><div id="pending" class="empty">admin login required</div></section>
<section data-tab="billing" data-admin-only><h2>Settlement history</h2><div id="settlements" class="empty">admin login required</div></section>
<section data-tab="admin" id="admin-section"><h2>All accounts (admin)</h2><div id="admin" class="empty"></div></section>
<section data-tab="admin" data-admin-only><h2>Strikes / bans / forfeitures</h2><div id="fraud" class="empty">admin login required</div></section>
<section data-tab="admin"><h2>Client balances</h2><div id="clients" class="empty">admin login required</div></section>
<section data-tab="admin" class="wide"><h2>Console output</h2><div id="console" class="console empty">admin login required</div></section>
</main> </main>
<script> <script>
"use strict"; "use strict";
@@ -211,7 +277,7 @@ function renderStats(stats) {
$("stats").innerHTML = table(["model", "rpm (1h)", "rpm (24h)", "rpm (30d)"], rows); $("stats").innerHTML = table(["model", "rpm (1h)", "rpm (24h)", "rpm (30d)"], rows);
} }
function renderThroughput(stats) { function renderThroughputHtml(stats) {
const nodes = (stats && stats.nodes) || {}; const nodes = (stats && stats.nodes) || {};
const rows = []; const rows = [];
for (const [nodeId, nodeStats] of Object.entries(nodes)) { for (const [nodeId, nodeStats] of Object.entries(nodes)) {
@@ -224,73 +290,266 @@ function renderThroughput(stats) {
]); ]);
} }
} }
$("throughput").innerHTML = table(["node", "model", "tps (1h)", "samples"], rows); if (!rows.length) return '<div class="empty">no throughput samples yet</div>';
return table(["node", "model", "tps (1h)", "samples"], rows);
} }
function renderInferenceHistory(data) { function hiveThroughputSummary(stats) {
const events = (data && data.events) || []; const nodes = (stats && stats.nodes) || {};
const started = new Map(); let totalTps = 0;
const progress = new Map(); let samples = 0;
const completed = []; for (const nodeStats of Object.values(nodes)) {
for (const e of events) { for (const s of Object.values((nodeStats && nodeStats.models) || {})) {
const f = e.fields || {}; const t = Number(s.tokens_per_sec_last_hour);
const id = f.request_id; if (Number.isFinite(t)) totalTps += t;
if (!id) continue; samples += Number(s.sample_count_last_hour || 0);
if (e.message === "proxy route selected") { }
started.set(id, e); }
} else if (e.message === "proxy progress") { return { totalTps, samples };
progress.set(id, e); }
} else if (e.message === "proxy complete" || e.message === "proxy failed" || e.message === "direct proxy failed after relay") {
completed.push(e); function buildCallWallStates(events) {
started.delete(id); const byId = new Map();
progress.delete(id); for (const e of events) {
}
}
const activeByModel = {};
let queuedEstimate = 0;
const activeRows = [];
for (const e of started.values()) {
const f = e.fields || {};
const model = f.model || f.route_model || "?";
activeByModel[model] = (activeByModel[model] || 0) + 1;
const p = (progress.get(f.request_id) || {}).fields || {};
const nodeQueues = Array.isArray(f.nodes) ? f.nodes.map(n => Number(n.queue_depth || 0)) : [];
const maxQueue = nodeQueues.length ? Math.max(...nodeQueues) : 0;
queuedEstimate += Math.max(0, maxQueue - 1);
activeRows.push([
new Date((e.ts || 0) * 1000).toLocaleTimeString(),
esc(short(model, 28)),
esc(short(f.request_id || "?", 18)),
`<span class="num">${esc(tps(p.tokens_per_sec))}</span>`,
`<span class="num">${esc(String(p.tokens ?? 0))}</span>`,
`<span class="num">${esc(String(maxQueue))}</span>`,
p.stream ? "stream" : "json",
]);
}
const active = Object.entries(activeByModel)
.map(([model, count]) => `${esc(model)}: <span class="num">${count}</span> active`)
.join(" &middot; ");
const liveSummary = active
? `${active}${queuedEstimate ? ` &middot; queued estimate: <span class="num">${queuedEstimate}</span>` : ""}`
: "no active requests";
const rows = completed.slice(-20).reverse().map(e => {
const f = e.fields || {}; const f = e.fields || {};
const id = f.request_id;
if (!id) continue;
let rec = byId.get(id);
if (!rec) {
rec = { id, events: [] };
byId.set(id, rec);
}
rec.events.push(e);
const msg = e.message;
if (msg === "proxy route selected") {
rec.status = "pending";
rec.started = e.ts;
rec.model = f.model || f.route_model || "?";
rec.route = f.route || f.nodes;
rec.nodes = f.nodes;
rec.stream = f.stream;
} else if (msg === "proxy via relay" || msg === "proxy connected") {
rec.status = "processing";
if (!rec.started) rec.started = e.ts;
rec.model = rec.model || f.model || f.route_model || "?";
} else if (msg === "proxy progress") {
rec.status = "processing";
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.stream = f.stream;
} else if (msg === "relay proxy failed, trying direct") {
rec.status = "processing";
rec.warn = "relay failed, trying direct";
} else if (msg === "proxy complete") {
rec.status = "complete";
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.stream = f.stream;
rec.terminal = e;
} else if (msg === "proxy failed" || msg === "direct proxy failed after relay") {
rec.status = "failed";
rec.model = rec.model || f.model || f.route_model || "?";
rec.error = f.error || msg;
rec.terminal = e;
}
}
return byId;
}
function callWallAgeSeconds(rec, nowSec) {
const start = rec.started || (rec.events[0] && rec.events[0].ts) || nowSec;
return Math.max(0, nowSec - start);
}
function callWallMaxQueue(rec) {
const nodes = rec.nodes || [];
const nodeQueues = Array.isArray(nodes) ? nodes.map(n => Number(n.queue_depth || 0)) : [];
return nodeQueues.length ? Math.max(...nodeQueues) : 0;
}
function renderCallWall(consoleData, stats) {
const events = (consoleData && consoleData.events) || [];
const nowSec = Date.now() / 1000;
const states = buildCallWallStates(events);
const active = [];
const terminal = [];
for (const rec of states.values()) {
if (rec.status === "pending" || rec.status === "processing") active.push(rec);
else if (rec.status === "complete" || rec.status === "failed") terminal.push(rec);
}
active.sort((a, b) => (a.started || 0) - (b.started || 0));
terminal.sort((a, b) => (b.terminal && b.terminal.ts) - (a.terminal && a.terminal.ts));
const hive = hiveThroughputSummary(stats);
const pending = active.filter(r => r.status === "pending").length;
const processing = active.filter(r => r.status === "processing").length;
const failedRecent = terminal.filter(r => r.status === "failed").length;
let queuedEstimate = 0;
for (const rec of active) queuedEstimate += Math.max(0, callWallMaxQueue(rec) - 1);
let html =
`<div class="dim" style="margin-bottom:6px">` +
`hive tps (1h): <b>${esc(tps(hive.totalTps))}</b> · samples: <b>${hive.samples}</b> · ` +
`active: <span class="status-processing">${processing}</span> processing · ` +
`<span class="status-pending">${pending}</span> pending` +
(queuedEstimate ? ` · queued estimate: <b>${queuedEstimate}</b>` : "") +
(failedRecent ? ` · <span class="status-failed">${failedRecent} recent failures</span>` : "") +
`</div>`;
if (active.length) {
html += table(["status", "age", "model", "request", "live tps", "tokens", "queue", "route / note"], active.map(rec => {
const statusCls = rec.status === "pending" ? "status-pending" : "status-processing";
const note = rec.warn || (rec.route ? short(String(rec.route), 28) : "");
return [
`<span class="${statusCls}">${esc(rec.status)}</span>`,
`<span class="num">${esc(callWallAgeSeconds(rec, nowSec).toFixed(1))}s</span>`,
esc(short(rec.model || "?", 28)),
esc(short(rec.id, 18)),
`<span class="num">${esc(tps(rec.tps))}</span>`,
`<span class="num">${esc(String(rec.tokens ?? "—"))}</span>`,
`<span class="num">${esc(String(callWallMaxQueue(rec)))}</span>`,
esc(note),
];
}));
} else {
html += '<div class="empty">no in-flight requests</div>';
}
const historyRows = terminal.slice(0, 40).map(rec => {
const e = rec.terminal || {};
const f = e.fields || {};
const statusCls = rec.status === "failed" ? "status-failed" : "status-complete";
const detail = rec.status === "failed"
? esc(short(rec.error || "?", 40))
: (f.stream ? "stream" : "json");
return [ return [
new Date((e.ts || 0) * 1000).toLocaleTimeString(), new Date((e.ts || 0) * 1000).toLocaleTimeString(),
esc(short(f.model || f.route_model || "?", 28)), `<span class="${statusCls}">${esc(rec.status)}</span>`,
esc(short(f.request_id || "?", 18)), esc(short(rec.model || "?", 28)),
`<span class="num">${esc(tps(f.tokens_per_sec))}</span>`, esc(short(rec.id, 18)),
`<span class="num">${esc(String(f.tokens ?? "?"))}</span>`, `<span class="num">${esc(tps(rec.tps ?? f.tokens_per_sec))}</span>`,
`<span class="num">${esc(String(f.elapsed_seconds ?? "?"))}</span>`, `<span class="num">${esc(String(rec.tokens ?? f.tokens ?? "?"))}</span>`,
f.stream ? "stream" : "json", `<span class="num">${esc(String(rec.elapsed ?? f.elapsed_seconds ?? "?"))}</span>`,
detail,
]; ];
}); });
$("inference-history").innerHTML = html += '<div style="margin-top:8px"><b class="dim">recent completed / failed</b></div>';
`<div class="dim" style="margin-bottom:6px">${liveSummary}</div>` + html += historyRows.length
(activeRows.length ? table(["started", "model", "request", "live tps", "tokens", "queue", "mode"], activeRows.reverse()) : "") + ? table(["time", "status", "model", "request", "tps", "tokens", "sec", "detail"], historyRows)
(rows.length ? table(["time", "model", "request", "tps", "tokens", "sec", "mode"], rows) : '<div class="empty">no completed requests yet</div>';
: '<div class="empty">no completed inference requests</div>'); $("call-wall").innerHTML = html;
} }
function startOfLocalDay(tsSec) {
const d = new Date(tsSec * 1000);
d.setHours(0, 0, 0, 0);
return d.getTime() / 1000;
}
function formatUsageDayLabel(tsSec) {
return new Date(tsSec * 1000).toLocaleDateString();
}
function summarizeUsageBuckets(records) {
const now = Date.now() / 1000;
const todayStart = startOfLocalDay(now);
const daySec = 86400;
const empty = () => ({ requests: 0, tokens: 0, cost: 0 });
const daily = [0, 1, 2].map(offset => ({
label: offset === 0 ? "Today" : offset === 1 ? "Yesterday" : formatUsageDayLabel(todayStart - offset * daySec),
...empty(),
}));
const last7 = { label: "Last 7 days", ...empty() };
const last30 = { label: "Last 30 days", ...empty() };
const total = { label: "All time", ...empty() };
for (const u of records) {
const ts = Number(u.ts || 0);
const tokens = Number(u.total_tokens || 0);
const cost = Number(u.cost || 0);
total.requests += 1;
total.tokens += tokens;
total.cost += cost;
if (ts >= now - 30 * daySec) {
last30.requests += 1;
last30.tokens += tokens;
last30.cost += cost;
}
if (ts >= now - 7 * daySec) {
last7.requests += 1;
last7.tokens += tokens;
last7.cost += cost;
}
for (let offset = 0; offset < 3; offset++) {
const start = todayStart - offset * daySec;
const end = start + daySec;
if (ts >= start && ts < end) {
daily[offset].requests += 1;
daily[offset].tokens += tokens;
daily[offset].cost += cost;
}
}
}
return [...daily, last7, last30, total];
}
function renderUsageSummary(records) {
const el = $("usage-summary");
if (!el) return;
if (!sessionToken) {
el.innerHTML = '<div class="empty">login required</div>';
return;
}
if (!records.length) {
el.innerHTML = '<div class="empty">no billed requests yet</div>';
return;
}
const rows = summarizeUsageBuckets(records).map(b => [
esc(b.label),
`<span class="num">${b.requests}</span>`,
`<span class="num">${esc(String(b.tokens))}</span>`,
`<span class="num">${usdt(b.cost)}</span>`,
]);
el.innerHTML =
'<div class="dim" style="margin-bottom:6px">per-request detail on Request history below</div>' +
table(["period", "requests", "tokens", "cost (USDT)"], rows);
}
function renderNodeThroughput(stats) {
const el = $("node-throughput");
if (!el) return;
if (!sessionToken) {
el.innerHTML = '<div class="empty">login required</div>';
return;
}
el.innerHTML = renderThroughputHtml(stats);
}
function renderBillingUsage(records) {
const el = $("billing-usage");
if (!el) return;
if (!sessionToken) {
el.innerHTML = '<div class="empty">login required</div>';
return;
}
if (!records.length) {
el.innerHTML = '<div class="empty">no billed requests yet</div>';
return;
}
const rows = records.slice().reverse().map(u => [
new Date((u.ts || 0) * 1000).toLocaleString(),
esc(short(u.model || "?", 28)),
esc(short(u.api_key || "?", 14)),
`<span class="num">${esc(String(u.total_tokens))}</span>`,
`<span class="num">${usdt(u.cost)}</span>`,
]);
el.innerHTML = `<div class="dim" style="margin-bottom:6px">${records.length} request${records.length === 1 ? "" : "s"}</div>` +
table(["time", "model", "api key", "tokens", "cost (USDT)"], rows);
}
function renderConsole(data) { function renderConsole(data) {
const events = (data && data.events) || []; const events = (data && data.events) || [];
@@ -311,10 +570,126 @@ function renderConsole(data) {
let sessionToken = localStorage.getItem("meshnet_session") || null; let sessionToken = localStorage.getItem("meshnet_session") || null;
let authTab = "login"; let authTab = "login";
let dashboardTab = "overview";
let isAdmin = false;
let isLoggedIn = false;
let accountApiKeys = [];
let accountUsageRecords = [];
let lastStats = null;
let availableModels = [];
let chatHistory = [];
let chatBusy = false;
let selectedChatModel = localStorage.getItem("meshnet_chat_model") || "";
async function apiCall(path, method, body) { function switchDashboardTab(name) {
if (name === "admin" && !isAdmin) name = "overview";
if (name === "billing" && !isLoggedIn) name = "overview";
dashboardTab = name;
updateSectionVisibility();
for (const tabName of ["overview", "chat", "billing", "admin"]) {
const button = $("tab-" + tabName);
if (button) button.classList.toggle("active", tabName === dashboardTab);
}
}
function updateSectionVisibility() {
for (const section of document.querySelectorAll("main section[data-tab]")) {
const onTab = section.dataset.tab === dashboardTab;
const adminOnly = section.hasAttribute("data-admin-only");
const loggedInOnly = section.hasAttribute("data-logged-in-only");
section.hidden = !onTab || (adminOnly && !isAdmin) || (loggedInOnly && !isLoggedIn);
}
}
function renderChatStatus(text) {
$("chat-status").textContent = text;
}
function renderChatHistory() {
const history = $("chat-history");
if (!chatHistory.length) {
history.classList.add("empty");
history.innerHTML = "no messages yet";
return;
}
history.classList.remove("empty");
history.innerHTML = chatHistory.map(msg => {
const roleClass = msg.role === "user" ? "chat-role-user" : msg.role === "assistant" ? "chat-role-assistant" : "chat-role-error";
const label = msg.role === "user" ? "user" : msg.role === "assistant" ? "assistant" : "error";
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("");
history.scrollTop = history.scrollHeight;
}
function renderChatModels() {
const select = $("chat-model");
if (!select) return;
const models = availableModels.slice();
if (!models.length) {
select.innerHTML = '<option value="">no models available</option>';
select.disabled = true;
return;
}
select.disabled = false;
const preferred = models.find(m => m.id === selectedChatModel)
|| models[0];
selectedChatModel = preferred.id;
localStorage.setItem("meshnet_chat_model", selectedChatModel);
select.innerHTML = models.map(model => {
const label = model.name && model.name !== model.id
? `${model.name} (${model.id})`
: model.id;
const suffix = model.recommended ? " [recommended]" : "";
return `<option value="${esc(model.id)}"${model.id === selectedChatModel ? " selected" : ""}>${esc(label + suffix)}</option>`;
}).join("");
select.value = selectedChatModel;
}
function selectChatModel(value) {
selectedChatModel = value || "";
localStorage.setItem("meshnet_chat_model", selectedChatModel);
}
function clearChatHistory() {
chatHistory = [];
renderChatHistory();
renderChatStatus("history cleared");
}
function chatAuthToken() {
if (accountApiKeys.length) return accountApiKeys[0];
return null;
}
function setAdminMode(enabled) {
isAdmin = enabled;
$("tab-admin").style.display = enabled ? "" : "none";
if (!enabled && dashboardTab === "admin") {
switchDashboardTab("overview");
} else {
updateSectionVisibility();
}
}
function setLoggedInMode(enabled) {
isLoggedIn = enabled;
$("tab-billing").style.display = enabled ? "" : "none";
if (!enabled) {
accountUsageRecords = [];
renderBillingUsage([]);
renderUsageSummary([]);
renderNodeThroughput(null);
if (dashboardTab === "billing") switchDashboardTab("overview");
} else {
updateSectionVisibility();
}
}
async function apiCall(path, method, body, bearerToken) {
const headers = { "Content-Type": "application/json" }; const headers = { "Content-Type": "application/json" };
if (sessionToken) headers["Authorization"] = "Bearer " + sessionToken; const token = bearerToken === undefined ? sessionToken : bearerToken;
if (token) headers["Authorization"] = "Bearer " + token;
try { try {
const r = await fetch(path, { const r = await fetch(path, {
method: method || "GET", method: method || "GET",
@@ -350,7 +725,10 @@ function renderAuthForms(errorMsg) {
$("account").innerHTML = $("account").innerHTML =
`<div class="tabs">${tab("login", "Log in")}${tab("register", "Register")}</div>` + `<div class="tabs">${tab("login", "Log in")}${tab("register", "Register")}</div>` +
form + `<div class="error-msg">${errorMsg ? esc(errorMsg) : ""}</div>`; form + `<div class="error-msg">${errorMsg ? esc(errorMsg) : ""}</div>`;
$("admin-section").style.display = "none"; accountApiKeys = [];
renderChatAuthHint();
setLoggedInMode(false);
setAdminMode(false);
} }
function switchAuthTab(name) { authTab = name; renderAuthForms(); } function switchAuthTab(name) { authTab = name; renderAuthForms(); }
@@ -398,15 +776,82 @@ async function topupKey(key) {
await renderAccountPanel(); await renderAccountPanel();
} }
const COPY_TOOLTIP_MS = 2000;
function showCopiedTooltip(anchor) {
const box = (anchor && anchor.closest && anchor.closest(".keybox")) || anchor;
if (!box) return;
const existing = box.querySelector(".copy-tooltip");
if (existing) existing.remove();
const tip = document.createElement("span");
tip.className = "copy-tooltip";
tip.textContent = "Copied!";
tip.setAttribute("role", "status");
box.appendChild(tip);
setTimeout(() => tip.remove(), COPY_TOOLTIP_MS);
}
async function copyApiKeyText(text, anchor) {
try {
await navigator.clipboard.writeText(text);
} catch {
const ta = document.createElement("textarea");
ta.value = text;
ta.style.position = "fixed";
ta.style.left = "-9999px";
document.body.appendChild(ta);
ta.select();
try { document.execCommand("copy"); } catch { /* ignore */ }
document.body.removeChild(ta);
}
if (anchor) showCopiedTooltip(anchor);
}
function selectApiKeyText(el) {
const range = document.createRange();
range.selectNodeContents(el);
const sel = window.getSelection();
if (!sel) return;
sel.removeAllRanges();
sel.addRange(range);
}
function copyApiKeyFromTextEl(el) {
const key = el.dataset.key || el.textContent || "";
return copyApiKeyText(key, el);
}
function copyApiKeyFromButton(button) {
const el = button.closest(".keybox") && button.closest(".keybox").querySelector(".key-text");
const key = (el && el.dataset.key) || "";
return copyApiKeyText(key, button);
}
function renderChatAuthHint() {
if (chatAuthToken()) {
renderChatStatus("ready to send with your active API key");
} else if (sessionToken) {
renderChatStatus("create an API key in Account to use chat on a billing-enabled tracker");
} else {
renderChatStatus("log in if this tracker requires an API key");
}
}
async function renderAccountPanel() { async function renderAccountPanel() {
const r = await apiCall("/v1/account"); const r = await apiCall("/v1/account");
if (r.status === 404) { // accounts disabled on this tracker if (r.status === 404) { // accounts disabled on this tracker
$("account-section").style.display = "none"; $("account-section").style.display = "none";
$("admin-section").style.display = "none"; accountApiKeys = [];
accountUsageRecords = [];
renderChatAuthHint();
setLoggedInMode(false);
setAdminMode(false);
return; return;
} }
if (!r.ok) { setSession(null); renderAuthForms(); return; } if (!r.ok) { setSession(null); renderAuthForms(); return; }
const { account, api_keys, balances, total_balance, usage, topup_amount } = r.data; const { account, api_keys, balances, total_balance, usage, topup_amount } = r.data;
accountApiKeys = Array.isArray(api_keys) ? api_keys.slice() : [];
accountUsageRecords = (usage && (usage.records || usage.recent)) || [];
const who = account.email || account.wallet || account.account_id; const who = account.email || account.wallet || account.account_id;
let html = let html =
`<div><b>${esc(who)}</b> <span class="pill">${esc(account.role)}</span> ` + `<div><b>${esc(who)}</b> <span class="pill">${esc(account.role)}</span> ` +
@@ -418,8 +863,10 @@ async function renderAccountPanel() {
'<button class="small" onclick="createKey()">+ new key</button></div>'; '<button class="small" onclick="createKey()">+ new key</button></div>';
if (api_keys.length) { if (api_keys.length) {
for (const key of api_keys) { for (const key of api_keys) {
html += `<div class="keybox">${esc(key)}` + html += `<div class="keybox">` +
` <span class="dim">(${usdt(balances[key] ?? 0)} USDT)</span>` + `<span class="key-text" data-key="${esc(key)}" onclick="selectApiKeyText(this)" ondblclick="copyApiKeyFromTextEl(this)">${esc(key)}</span>` +
`<span class="dim">(${usdt(balances[key] ?? 0)} USDT)</span>` +
`<button class="small" type="button" onclick="copyApiKeyFromButton(this)">copy</button>` +
(topup_amount > 0 (topup_amount > 0
? ` <button class="small" onclick="topupKey('${esc(key)}')">+${usdt(topup_amount)} (devnet)</button>` ? ` <button class="small" onclick="topupKey('${esc(key)}')">+${usdt(topup_amount)} (devnet)</button>`
: "") + : "") +
@@ -428,24 +875,74 @@ async function renderAccountPanel() {
} else { } else {
html += '<div class="empty">no active keys</div>'; html += '<div class="empty">no active keys</div>';
} }
if (usage.recent && usage.recent.length) {
html += '<div style="margin-top:6px"><b class="dim">recent usage</b></div>' +
table(["time", "model", "tokens", "cost"], usage.recent.slice().reverse().map(u => [
new Date(u.ts * 1000).toLocaleTimeString(),
esc(short(u.model || "?", 24)),
`<span class="num">${esc(String(u.total_tokens))}</span>`,
`<span class="num">${usdt(u.cost)}</span>`,
]));
}
$("account").innerHTML = html; $("account").innerHTML = html;
renderUsageSummary(accountUsageRecords);
renderNodeThroughput(lastStats);
renderBillingUsage(accountUsageRecords);
renderChatAuthHint();
renderChatModels();
renderChatHistory();
setLoggedInMode(true);
setAdminMode(account.role === "admin");
if (account.role === "admin") await renderAdminPanel(); if (account.role === "admin") await renderAdminPanel();
else $("admin-section").style.display = "none"; }
async function sendChat() {
const promptEl = $("chat-prompt");
const prompt = promptEl.value.trim();
if (!prompt || chatBusy) return;
if (!selectedChatModel) {
renderChatStatus("select a model first");
return;
}
const bearerToken = chatAuthToken();
const body = {
model: selectedChatModel,
messages: [
...chatHistory
.filter(msg => msg.role === "user" || msg.role === "assistant")
.map(msg => ({ role: msg.role, content: msg.content })),
{ role: "user", content: prompt },
],
stream: false,
max_tokens: 256,
};
chatBusy = true;
$("chat-send").disabled = true;
promptEl.value = "";
chatHistory.push({ role: "user", content: prompt, model: selectedChatModel });
renderChatHistory();
renderChatStatus("sending request…");
const r = await apiCall("/v1/chat/completions", "POST", body, bearerToken);
chatBusy = false;
$("chat-send").disabled = false;
if (!r.ok) {
const error = r.data && r.data.error
? (typeof r.data.error === "string" ? r.data.error : r.data.error.message || "request failed")
: "request failed";
chatHistory.push({ role: "error", content: error, model: selectedChatModel });
renderChatHistory();
renderChatStatus(error);
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({
role: "assistant",
content: reply || "(empty response)",
model: selectedChatModel,
});
renderChatHistory();
renderChatStatus(usage
? `done: ${usage.total_tokens ?? "?"} tokens`
: "done");
promptEl.focus();
} }
async function renderAdminPanel() { async function renderAdminPanel() {
const r = await apiCall("/v1/admin/accounts"); const r = await apiCall("/v1/admin/accounts");
if (!r.ok) { $("admin-section").style.display = "none"; return; } if (!r.ok) { setAdminMode(false); return; }
$("admin-section").style.display = "";
const rows = (r.data.accounts || []).map(a => { const rows = (r.data.accounts || []).map(a => {
const balance = Object.values(a.balances || {}).reduce((x, y) => x + y, 0); const balance = Object.values(a.balances || {}).reduce((x, y) => x + y, 0);
return [ return [
@@ -461,28 +958,44 @@ async function renderAdminPanel() {
async function refresh() { async function refresh() {
$("self-url").textContent = location.host; $("self-url").textContent = location.host;
const [raft, map, summary, settlements, wallets, stats, consoleData] = await Promise.all([ const [raft, map, stats, models, consoleData, adminData] = await Promise.all([
fetchJson("/v1/raft/status"), fetchJson("/v1/raft/status"),
fetchJson("/v1/network/map"), fetchJson("/v1/network/map"),
fetchJson("/v1/billing/summary"),
fetchJson("/v1/billing/settlements"),
fetchJson("/v1/registry/wallets"),
fetchJson("/v1/stats"), fetchJson("/v1/stats"),
fetchJson("/v1/models"),
fetchJson("/v1/console"), fetchJson("/v1/console"),
isAdmin ? Promise.all([
fetchJson("/v1/billing/summary"),
fetchJson("/v1/billing/settlements"),
fetchJson("/v1/registry/wallets"),
]) : Promise.resolve([null, null, null]),
]); ]);
const [summary, settlements, wallets] = adminData;
lastStats = stats;
availableModels = ((models && models.data) || []).map(model => ({
id: model.id,
name: model.name || model.id,
recommended: Boolean(model.recommended),
aliases: model.aliases || [],
})).filter(model => model.id);
renderHive(raft); renderHive(raft);
renderNodes(map); renderNodes(map);
renderBilling(summary); renderBilling(summary);
renderSettlements(settlements); renderSettlements(settlements);
renderFraud(wallets, summary); renderFraud(wallets, summary);
renderStats(stats); renderStats(stats);
renderThroughput(stats); renderCallWall(consoleData, stats);
renderInferenceHistory(consoleData);
renderConsole(consoleData); renderConsole(consoleData);
renderNodeThroughput(stats);
renderChatModels();
renderChatHistory();
$("refreshed").textContent = "refreshed " + new Date().toLocaleTimeString(); $("refreshed").textContent = "refreshed " + new Date().toLocaleTimeString();
} }
refresh(); refresh();
renderAccountPanel(); renderAccountPanel();
renderChatModels();
renderChatHistory();
renderChatAuthHint();
setInterval(refresh, 4000); setInterval(refresh, 4000);
setInterval(() => { if (sessionToken) renderAccountPanel(); }, 8000); setInterval(() => { if (sessionToken) renderAccountPanel(); }, 8000);
</script> </script>

738
packages/tracker/meshnet_tracker/server.py
View File

@@ -142,18 +142,18 @@ DEFAULT_CALLER_CREDIT_USDT = 1.0
DEFAULT_DEVNET_TOPUP_USDT = 1.0 DEFAULT_DEVNET_TOPUP_USDT = 1.0
def _model_aliases(model: str | None) -> set[str]: def _model_aliases(model: str | None) -> set[str]:
"""Return stable lookup aliases for a model repo or display name.""" """Return stable lookup aliases for a model repo or display name."""
if not model: if not model:
return set() return set()
normalized = model.strip() normalized = model.strip()
if not normalized: if not normalized:
return set() return set()
aliases = {normalized} aliases = {normalized}
short = normalized.rsplit("/", 1)[-1] short = normalized.rsplit("/", 1)[-1]
aliases.add(short) aliases.add(short)
lowered = short.lower() lowered = short.lower()
aliases.add(lowered) aliases.add(lowered)
if lowered.endswith("-instruct"): if lowered.endswith("-instruct"):
aliases.add(lowered.removesuffix("-instruct")) aliases.add(lowered.removesuffix("-instruct"))
return aliases return aliases
@@ -542,8 +542,9 @@ class _NodeEntry:
"model_tokens_per_sec", "model_tokens_per_sec",
"pending_directives", "last_heartbeat", "tracker_mode", "pending_directives", "last_heartbeat", "tracker_mode",
"relay_addr", "cert_fingerprint", "peer_id", "relay_addr", "cert_fingerprint", "peer_id",
# heartbeat stats (reported by node, cumulative) # heartbeat stats (reported by node, cumulative)
"total_requests", "failed_requests", "queue_depth", "proxy_inflight", "uptime_seconds", "total_requests", "failed_requests", "queue_depth", "proxy_inflight", "uptime_seconds",
"current_requests",
"status", # "ready" | "loading" "status", # "ready" | "loading"
"heartbeats_expected", "heartbeats_received", "heartbeats_expected", "heartbeats_received",
# dynamic reassignment queued by the tracker # dynamic reassignment queued by the tracker
@@ -604,18 +605,19 @@ class _NodeEntry:
self.peer_id = peer_id self.peer_id = peer_id
self.pending_directives: list[dict] = [] self.pending_directives: list[dict] = []
self.last_heartbeat: float = time.monotonic() self.last_heartbeat: float = time.monotonic()
self.total_requests: int = 0 self.total_requests: int = 0
self.failed_requests: int = 0 self.failed_requests: int = 0
self.queue_depth: int = 0 self.queue_depth: int = 0
self.proxy_inflight: int = 0 self.proxy_inflight: int = 0
self.uptime_seconds: float = 0.0 self.current_requests: list[dict] = []
self.uptime_seconds: float = 0.0
self.status: str = "ready" self.status: str = "ready"
self.heartbeats_expected: int = 0 self.heartbeats_expected: int = 0
self.heartbeats_received: int = 0 self.heartbeats_received: int = 0
self.pending_new_assignment: dict | None = None self.pending_new_assignment: dict | None = None
def _effective_throughput(node: "_NodeEntry", model: str | None = None) -> float: def _effective_throughput(node: "_NodeEntry", model: str | None = None) -> float:
"""Effective tokens/s accounting for current queue depth.""" """Effective tokens/s accounting for current queue depth."""
observed = None observed = None
if model: if model:
@@ -626,22 +628,56 @@ def _effective_throughput(node: "_NodeEntry", model: str | None = None) -> float
if observed is not None: if observed is not None:
break break
base = observed if observed is not None and observed > 0 else node.benchmark_tokens_per_sec base = observed if observed is not None and observed > 0 else node.benchmark_tokens_per_sec
return base / (_effective_queue_depth(node) + 1) return base / (_effective_queue_depth(node) + 1)
def _effective_queue_depth(node: "_NodeEntry") -> int: def _effective_queue_depth(node: "_NodeEntry") -> int:
"""Best current load estimate: heartbeat queue or tracker-routed in-flight.""" """Best current load estimate: heartbeat queue or tracker-routed in-flight."""
return max(node.queue_depth, node.proxy_inflight) return max(node.queue_depth, node.proxy_inflight)
def _record_proxy_inflight( _CURRENT_REQUEST_FIELDS = frozenset({
server: "_TrackerHTTPServer", "request_id", "model", "kind", "tokens", "tokens_per_sec",
nodes: list["_NodeEntry"], "elapsed_seconds", "routing_complete",
delta: int, })
) -> None:
with server.lock:
for node in nodes: def _normalize_current_requests(items: object, *, limit: int = 32) -> list[dict]:
node.proxy_inflight = max(0, node.proxy_inflight + delta) """Sanitize node-reported in-flight request snapshots from heartbeats."""
if not isinstance(items, list):
return []
out: list[dict] = []
for item in items[:limit]:
if not isinstance(item, dict):
continue
request_id = item.get("request_id")
if not request_id:
continue
rec: dict = {"request_id": str(request_id)}
for key in _CURRENT_REQUEST_FIELDS:
if key == "request_id" or key not in item:
continue
value = item[key]
if key in {"tokens"}:
rec[key] = int(value)
elif key in {"tokens_per_sec", "elapsed_seconds"}:
rec[key] = float(value)
elif key == "routing_complete":
rec[key] = bool(value)
else:
rec[key] = str(value)
out.append(rec)
return out
def _record_proxy_inflight(
server: "_TrackerHTTPServer",
nodes: list["_NodeEntry"],
delta: int,
) -> None:
with server.lock:
for node in nodes:
node.proxy_inflight = max(0, node.proxy_inflight + delta)
def _reputation_multiplier(node: "_NodeEntry", contracts: Any | None) -> float: def _reputation_multiplier(node: "_NodeEntry", contracts: Any | None) -> float:
@@ -707,12 +743,13 @@ def _node_route_summary(nodes: list["_NodeEntry"]) -> list[dict]:
"model": node.model, "model": node.model,
"hf_repo": node.hf_repo, "hf_repo": node.hf_repo,
"endpoint": node.endpoint, "endpoint": node.endpoint,
"shard": f"{node.shard_start}-{node.shard_end}", "shard": f"{node.shard_start}-{node.shard_end}",
"num_layers": node.num_layers, "num_layers": node.num_layers,
"queue_depth": _effective_queue_depth(node), "queue_depth": _effective_queue_depth(node),
"heartbeat_queue_depth": node.queue_depth, "heartbeat_queue_depth": node.queue_depth,
"proxy_inflight": node.proxy_inflight, "proxy_inflight": node.proxy_inflight,
} "current_requests": list(node.current_requests),
}
for node in nodes for node in nodes
] ]
@@ -749,11 +786,11 @@ def _coverage_percentage(
return round((covered / required_layers) * 100, 2) return round((covered / required_layers) * 100, 2)
def _served_model_copies( def _served_model_copies(
nodes: list[_NodeEntry], nodes: list[_NodeEntry],
required_start: int, required_start: int,
required_end: int, required_end: int,
) -> float: ) -> float:
required_layers = required_end - required_start + 1 required_layers = required_end - required_start + 1
if required_layers <= 0: if required_layers <= 0:
return 0.0 return 0.0
@@ -772,60 +809,60 @@ def _served_model_copies(
return 0.0 return 0.0
complete_copies = min(layer_counts) complete_copies = min(layer_counts)
residual_layers = sum(1 for count in layer_counts if count > complete_copies) residual_layers = sum(1 for count in layer_counts if count > complete_copies)
return round(complete_copies + (residual_layers / required_layers), 2) return round(complete_copies + (residual_layers / required_layers), 2)
def _model_health_summary( def _model_health_summary(
server: "_TrackerHTTPServer", server: "_TrackerHTTPServer",
model: str | None, model: str | None,
hf_repo: str | None = None, hf_repo: str | None = None,
) -> dict: ) -> dict:
lookup = hf_repo or model lookup = hf_repo or model
if not lookup: if not lookup:
return {} return {}
resolved_name, preset = _resolve_model_preset(server.model_presets, lookup) resolved_name, preset = _resolve_model_preset(server.model_presets, lookup)
if preset is not None: if preset is not None:
required_start, required_end = _preset_layer_bounds(preset) required_start, required_end = _preset_layer_bounds(preset)
model_nodes = [ model_nodes = [
node for node in server.registry.values() node for node in server.registry.values()
if _node_matches_preset(node, resolved_name, preset) # type: ignore[arg-type] if _node_matches_preset(node, resolved_name, preset) # type: ignore[arg-type]
] ]
model_id = resolved_name or lookup model_id = resolved_name or lookup
else: else:
model_nodes = [ model_nodes = [
node for node in server.registry.values() node for node in server.registry.values()
if _node_matches_model(node, lookup) if _node_matches_model(node, lookup)
and node.shard_start is not None and node.shard_start is not None
and node.shard_end is not None and node.shard_end is not None
and node.num_layers is not None and node.num_layers is not None
] ]
if not model_nodes: if not model_nodes:
return { return {
"model": lookup, "model": lookup,
"served_model_copies": 0.0, "served_model_copies": 0.0,
"coverage_percentage": 0.0, "coverage_percentage": 0.0,
"node_count": 0, "node_count": 0,
} }
required_start = 0 required_start = 0
required_end = max(node.num_layers for node in model_nodes if node.num_layers is not None) - 1 required_end = max(node.num_layers for node in model_nodes if node.num_layers is not None) - 1
model_id = hf_repo or model or lookup model_id = hf_repo or model or lookup
return { return {
"model": model_id, "model": model_id,
"required_start": required_start, "required_start": required_start,
"required_end": required_end, "required_end": required_end,
"served_model_copies": _served_model_copies(model_nodes, required_start, required_end), "served_model_copies": _served_model_copies(model_nodes, required_start, required_end),
"coverage_percentage": _coverage_percentage(model_nodes, required_start, required_end), "coverage_percentage": _coverage_percentage(model_nodes, required_start, required_end),
"node_count": len(model_nodes), "node_count": len(model_nodes),
} }
def _preset_layer_bounds(preset: dict) -> tuple[int, int]: def _preset_layer_bounds(preset: dict) -> tuple[int, int]:
start = int(preset.get("layers_start", 0)) start = int(preset.get("layers_start", 0))
if "layers_end" in preset: if "layers_end" in preset:
return start, int(preset["layers_end"]) return start, int(preset["layers_end"])
return start, start + int(preset["total_layers"]) - 1 return start, start + int(preset["total_layers"]) - 1
@@ -986,13 +1023,14 @@ def _node_health(node: "_NodeEntry", heartbeat_timeout: float) -> dict:
return { return {
"node_id": node.node_id, "node_id": node.node_id,
"endpoint": node.endpoint, "endpoint": node.endpoint,
"alive": alive, "alive": alive,
"last_seen_seconds_ago": round(age, 1), "last_seen_seconds_ago": round(age, 1),
"status": node.status, "status": node.status,
"queue_depth": _effective_queue_depth(node), "queue_depth": _effective_queue_depth(node),
"heartbeat_queue_depth": node.queue_depth, "heartbeat_queue_depth": node.queue_depth,
"proxy_inflight": node.proxy_inflight, "proxy_inflight": node.proxy_inflight,
"total_requests": node.total_requests, "current_requests": list(node.current_requests),
"total_requests": node.total_requests,
"heartbeat_success_rate": hb_rate, "heartbeat_success_rate": hb_rate,
"inference_success_rate": inf_rate, "inference_success_rate": inf_rate,
"capacity": _node_capacity_summary(node), "capacity": _node_capacity_summary(node),
@@ -1193,22 +1231,22 @@ def _billable_non_stream_split(payload: dict, request_body: dict) -> tuple[int,
Prefers the response usage block; falls back to content estimates. Prefers the response usage block; falls back to content estimates.
Completion stays capped by the request's max-tokens bound, as before. Completion stays capped by the request's max-tokens bound, as before.
""" """
usage = _usage_split(payload) usage = _usage_split(payload)
prompt_estimate = _estimate_prompt_tokens(request_body) or 0 prompt_estimate = _estimate_prompt_tokens(request_body) or 0
prompt = (usage or {}).get("prompt") prompt = (usage or {}).get("prompt")
completion = (usage or {}).get("completion") completion = (usage or {}).get("completion")
if prompt is None: if prompt is None:
prompt = prompt_estimate prompt = prompt_estimate
if completion is None: if completion is None:
total = (usage or {}).get("total") total = (usage or {}).get("total")
if total is not None: if total is not None:
completion = max(0, total - prompt) completion = max(0, total - prompt)
else: else:
completion = _observed_non_stream_completion_tokens(payload) completion = _observed_non_stream_completion_tokens(payload)
limit = _requested_completion_token_limit(request_body) limit = _requested_completion_token_limit(request_body)
if limit is not None and completion > limit: if limit is not None and completion > limit:
completion = min(completion, limit) completion = min(completion, limit)
prompt = max(prompt, prompt_estimate) prompt = max(prompt, prompt_estimate)
return max(0, prompt), max(0, completion) return max(0, prompt), max(0, completion)
@@ -1359,32 +1397,32 @@ def _drop_directive(node: _NodeEntry, model: str, start: int, end: int, quantiza
} }
def _purge_expired_nodes_locked(server: "_TrackerHTTPServer") -> list[str]: def _purge_expired_nodes_locked(server: "_TrackerHTTPServer") -> list[str]:
now = time.monotonic() now = time.monotonic()
expired_ids = [ expired_ids = [
node_id for node_id, entry in server.registry.items() node_id for node_id, entry in server.registry.items()
if (now - entry.last_heartbeat) > server.heartbeat_timeout if (now - entry.last_heartbeat) > server.heartbeat_timeout
] ]
expired_entries: list[tuple[str, _NodeEntry]] = [] expired_entries: list[tuple[str, _NodeEntry]] = []
for node_id in expired_ids: for node_id in expired_ids:
entry = server.registry.pop(node_id) entry = server.registry.pop(node_id)
expired_entries.append((node_id, entry)) expired_entries.append((node_id, entry))
if expired_ids: if expired_ids:
_rebalance_all_locked(server) _rebalance_all_locked(server)
for node_id, entry in expired_entries: for node_id, entry in expired_entries:
_tracker_log( _tracker_log(
server, server,
"warn", "warn",
"node expired", "node expired",
node_id=node_id, node_id=node_id,
endpoint=entry.endpoint, endpoint=entry.endpoint,
model=entry.model, model=entry.model,
hf_repo=entry.hf_repo, hf_repo=entry.hf_repo,
shard=f"{entry.shard_start}-{entry.shard_end}", shard=f"{entry.shard_start}-{entry.shard_end}",
heartbeat_timeout_seconds=server.heartbeat_timeout, heartbeat_timeout_seconds=server.heartbeat_timeout,
model_health=_model_health_summary(server, entry.model, entry.hf_repo), model_health=_model_health_summary(server, entry.model, entry.hf_repo),
) )
return expired_ids return expired_ids
def _rebalance_model_locked(server: "_TrackerHTTPServer", model: str) -> None: def _rebalance_model_locked(server: "_TrackerHTTPServer", model: str) -> None:
@@ -1747,7 +1785,7 @@ 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, **fields: Any) -> None:
event = { event = {
"ts": time.time(), "ts": time.time(),
"level": level, "level": level,
@@ -1761,41 +1799,41 @@ def _tracker_log(server: "_TrackerHTTPServer", level: str, message: str, **field
with server.console_lock: with server.console_lock:
server.console_events.append(event) 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) print(f"[tracker] {level}: {message}{suffix}", flush=True)
def _tracker_log_proxy_progress( def _tracker_log_proxy_progress(
server: "_TrackerHTTPServer", server: "_TrackerHTTPServer",
*, *,
request_id: str, request_id: str,
model: str, model: str,
route_model: str, route_model: str,
tokens: int, tokens: int,
started: float, started: float,
route_nodes: list["_NodeEntry"], route_nodes: list["_NodeEntry"],
stream: bool = True, stream: bool = True,
relay: bool = False, relay: bool = False,
) -> None: ) -> None:
elapsed = time.monotonic() - started elapsed = time.monotonic() - started
effective_elapsed = max(elapsed, 1e-6) effective_elapsed = max(elapsed, 1e-6)
_tracker_log( _tracker_log(
server, server,
"info", "info",
"proxy progress", "proxy progress",
request_id=request_id, request_id=request_id,
model=model, model=model,
route_model=route_model, route_model=route_model,
stream=stream, stream=stream,
relay=relay or None, relay=relay or None,
tokens=tokens, tokens=tokens,
elapsed_seconds=round(elapsed, 4), elapsed_seconds=round(elapsed, 4),
tokens_per_sec=round(tokens / effective_elapsed, 4) if tokens > 0 else 0.0, tokens_per_sec=round(tokens / effective_elapsed, 4) if tokens > 0 else 0.0,
route=_node_route_summary(route_nodes), route=_node_route_summary(route_nodes),
) )
def _node_id_for_registration( def _node_id_for_registration(
endpoint: str, endpoint: str,
model: str, model: str,
wallet_address: str | None, wallet_address: str | None,
@@ -2276,8 +2314,8 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
result[model] = server.stats.get_node_model_stats(node.node_id, model) result[model] = server.stats.get_node_model_stats(node.node_id, model)
return result return result
def model_supply_for(node: _NodeEntry) -> dict: def model_supply_for(node: _NodeEntry) -> dict:
return _model_health_summary(server, node.model, node.hf_repo) return _model_health_summary(server, node.model, node.hf_repo)
self._send_json(200, { self._send_json(200, {
"relay_url": server.relay_url, "relay_url": server.relay_url,
@@ -2479,6 +2517,8 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
node = max(candidates, key=lambda n: _effective_throughput(n, model)) node = max(candidates, key=lambda n: _effective_throughput(n, model))
target_url = f"{node.endpoint}/v1/chat/completions" target_url = f"{node.endpoint}/v1/chat/completions"
request_id = str(body.get("id") or f"req-{time.time_ns():x}") request_id = str(body.get("id") or f"req-{time.time_ns():x}")
body["id"] = request_id
raw_body = json.dumps(body).encode()
# Pre-resolve the downstream route so the first-shard node skips its own # Pre-resolve the downstream route so the first-shard node skips its own
# tracker query. We already hold the full registry picture — no need for # tracker query. We already hold the full registry picture — no need for
@@ -2540,23 +2580,23 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
# Strip the first-shard node we're about to proxy to — it's already handling the request. # Strip the first-shard node we're about to proxy to — it's already handling the request.
downstream_hops = [h for h in route_hops if h["endpoint"].rstrip("/") != node.endpoint.rstrip("/")] downstream_hops = [h for h in route_hops if h["endpoint"].rstrip("/") != node.endpoint.rstrip("/")]
downstream_urls = json.dumps(downstream_hops) downstream_urls = json.dumps(downstream_hops)
route_debug = " -> ".join( route_debug = " -> ".join(
f"{n.node_id}@{n.endpoint}[{n.shard_start}-{n.shard_end}]" f"{n.node_id}@{n.endpoint}[{n.shard_start}-{n.shard_end}]"
for n in route_nodes for n in route_nodes
) )
inflight_nodes = route_nodes or [node] inflight_nodes = route_nodes or [node]
inflight_recorded = True inflight_recorded = True
_record_proxy_inflight(server, inflight_nodes, 1) _record_proxy_inflight(server, inflight_nodes, 1)
def finish_proxy_inflight() -> None: def finish_proxy_inflight() -> None:
nonlocal inflight_recorded nonlocal inflight_recorded
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
_tracker_log( _tracker_log(
server, server,
"info", "info",
"proxy route selected", "proxy route selected",
request_id=request_id, request_id=request_id,
model=model, model=model,
@@ -2607,16 +2647,16 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
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.
self._stream_relayed_frames( self._stream_relayed_frames(
first, frames, started, first, frames, started,
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,
) )
finish_proxy_inflight() finish_proxy_inflight()
return return
if first is not None: if first is not None:
elapsed = time.monotonic() - started elapsed = time.monotonic() - started
self._send_relayed_response(first) self._send_relayed_response(first)
if int(first.get("status", 503)) < 400: if int(first.get("status", 503)) < 400:
body_text = first.get("body") or "" body_text = first.get("body") or ""
@@ -2639,12 +2679,12 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
tokens_per_sec=round(tokens / elapsed, 4) if elapsed > 0 else 0.0, tokens_per_sec=round(tokens / elapsed, 4) if elapsed > 0 else 0.0,
route=_node_route_summary(route_nodes), route=_node_route_summary(route_nodes),
) )
self._bill_completed( self._bill_completed(
api_key, model, tokens, node_work, api_key, model, tokens, node_work,
input_tokens=in_tokens, output_tokens=out_tokens, input_tokens=in_tokens, output_tokens=out_tokens,
) )
finish_proxy_inflight() finish_proxy_inflight()
return return
_tracker_log( _tracker_log(
server, server,
"warn", "warn",
@@ -2665,12 +2705,12 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
self.send_header("Content-Type", "application/json") self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(err_body))) self.send_header("Content-Length", str(len(err_body)))
self.end_headers() self.end_headers()
try: try:
self.wfile.write(err_body) self.wfile.write(err_body)
except BrokenPipeError: except BrokenPipeError:
pass pass
finish_proxy_inflight() finish_proxy_inflight()
return return
except Exception as exc: except Exception as exc:
if node.relay_addr: if node.relay_addr:
_tracker_log( _tracker_log(
@@ -2691,13 +2731,13 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
target_url=target_url, target_url=target_url,
error=repr(exc), error=repr(exc),
) )
self._send_json(503, {"error": { self._send_json(503, {"error": {
"message": f"upstream node unreachable: {exc}", "message": f"upstream node unreachable: {exc}",
"type": "service_unavailable", "type": "service_unavailable",
"code": "upstream_error", "code": "upstream_error",
}}) }})
finish_proxy_inflight() finish_proxy_inflight()
return return
with upstream: with upstream:
content_type = upstream.headers.get("Content-Type", "application/json") content_type = upstream.headers.get("Content-Type", "application/json")
@@ -2706,38 +2746,38 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
self.send_response(200) self.send_response(200)
self.send_header("Content-Type", "text/event-stream; charset=utf-8") self.send_header("Content-Type", "text/event-stream; charset=utf-8")
self.send_header("Cache-Control", "no-cache") self.send_header("Cache-Control", "no-cache")
self.end_headers() self.end_headers()
stream_usage: dict | None = None stream_usage: dict | None = None
observed_stream_tokens = 0 observed_stream_tokens = 0
client_gone = False client_gone = False
try: try:
while True: while True:
line = upstream.readline() line = upstream.readline()
if not line: if not line:
break break
if not client_gone: if not client_gone:
try: try:
self.wfile.write(line) self.wfile.write(line)
self.wfile.flush() self.wfile.flush()
except (BrokenPipeError, ConnectionResetError): except (BrokenPipeError, ConnectionResetError):
# Keep draining upstream so completed node work is still billed. # Keep draining upstream so completed node work is still billed.
client_gone = True client_gone = True
observed, usage = _stream_line_tokens(line) observed, usage = _stream_line_tokens(line)
observed_stream_tokens += observed observed_stream_tokens += observed
if observed: if observed:
_tracker_log_proxy_progress( _tracker_log_proxy_progress(
server, server,
request_id=request_id, request_id=request_id,
model=model, model=model,
route_model=route_model, route_model=route_model,
tokens=observed_stream_tokens, tokens=observed_stream_tokens,
started=started, started=started,
route_nodes=route_nodes, route_nodes=route_nodes,
) )
if usage is not None: if usage is not None:
stream_usage = usage stream_usage = usage
except (BrokenPipeError, ConnectionResetError): except (BrokenPipeError, ConnectionResetError):
pass pass
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;
@@ -2804,13 +2844,13 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
self.send_header("Content-Type", content_type) self.send_header("Content-Type", content_type)
self.send_header("Content-Length", str(len(resp_body))) self.send_header("Content-Length", str(len(resp_body)))
self.end_headers() self.end_headers()
try: try:
self.wfile.write(resp_body) self.wfile.write(resp_body)
except (BrokenPipeError, ConnectionResetError): except (BrokenPipeError, ConnectionResetError):
pass pass
finish_proxy_inflight() finish_proxy_inflight()
def _record_observed_throughput( def _record_observed_throughput(
self, self,
requested_model: str, requested_model: str,
route_model: str, route_model: str,
@@ -2823,12 +2863,12 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
The tracker sees end-to-end request duration, not per-hop timings, so The tracker sees end-to-end request duration, not per-hop timings, so
each hop gets the same route-level observation for now. Per-hop telemetry each hop gets the same route-level observation for now. Per-hop telemetry
can refine this later without changing the external stats shape. can refine this later without changing the external stats shape.
""" """
server: _TrackerHTTPServer = self.server # type: ignore[assignment] server: _TrackerHTTPServer = self.server # type: ignore[assignment]
if server.stats is None or total_tokens <= 0: if server.stats is None or total_tokens <= 0:
return return
elapsed_seconds = max(elapsed_seconds, 1e-6) elapsed_seconds = max(elapsed_seconds, 1e-6)
models = [m for m in (requested_model, route_model) if m] models = [m for m in (requested_model, route_model) if m]
if len(models) == 2 and models[0] == models[1]: if len(models) == 2 and models[0] == models[1]:
models = [models[0]] models = [models[0]]
for node in route_nodes: for node in route_nodes:
@@ -2935,21 +2975,21 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
model: str, model: str,
route_model: str, route_model: str,
route_nodes: list, route_nodes: list,
api_key: str | None, api_key: str | None,
node_work: list, node_work: list,
request_body: dict, request_body: dict,
request_id: str, request_id: str,
) -> 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."""
headers = first.get("headers") if isinstance(first.get("headers"), dict) else {} headers = first.get("headers") if isinstance(first.get("headers"), dict) else {}
self.send_response(int(first.get("status", 200))) self.send_response(int(first.get("status", 200)))
self.send_header("Content-Type", headers.get("Content-Type", "text/event-stream; charset=utf-8")) self.send_header("Content-Type", headers.get("Content-Type", "text/event-stream; charset=utf-8"))
self.send_header("Cache-Control", "no-cache") self.send_header("Cache-Control", "no-cache")
self.end_headers() self.end_headers()
server: _TrackerHTTPServer = self.server # type: ignore[assignment] server: _TrackerHTTPServer = self.server # type: ignore[assignment]
stream_usage: dict | None = None stream_usage: dict | None = None
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):
chunk = frame.get("chunk") or "" chunk = frame.get("chunk") or ""
@@ -2963,22 +3003,22 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
except BrokenPipeError: except BrokenPipeError:
# Keep draining frames — the nodes did the work; bill it. # Keep draining frames — the nodes did the work; bill it.
client_gone = True client_gone = True
for line in data.splitlines(): for line in data.splitlines():
observed, usage = _stream_line_tokens(line) observed, usage = _stream_line_tokens(line)
observed_stream_tokens += observed observed_stream_tokens += observed
if observed: if observed:
_tracker_log_proxy_progress( _tracker_log_proxy_progress(
server, server,
request_id=request_id, request_id=request_id,
model=model, model=model,
route_model=route_model, route_model=route_model,
tokens=observed_stream_tokens, tokens=observed_stream_tokens,
started=started, started=started,
route_nodes=route_nodes, route_nodes=route_nodes,
relay=True, relay=True,
) )
if usage is not None: if usage is not None:
stream_usage = usage stream_usage = usage
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
@@ -2987,11 +3027,11 @@ 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
_tracker_log( _tracker_log(
server, server,
"info", "info",
"proxy complete", "proxy complete",
request_id=request_id, request_id=request_id,
model=model, model=model,
route_model=route_model, route_model=route_model,
status=200, status=200,
@@ -3208,50 +3248,50 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
with server.lock: with server.lock:
self._purge_expired_nodes() self._purge_expired_nodes()
# Dedup: if this endpoint is already registered, remove the old entry first. # Dedup: if this endpoint is already registered, remove the old entry first.
stale_ids = [ stale_ids = [
eid for eid, e in server.registry.items() eid for eid, e in server.registry.items()
if e.endpoint == entry.endpoint.rstrip("/") if e.endpoint == entry.endpoint.rstrip("/")
] ]
stale_entries: list[tuple[str, _NodeEntry]] = [] stale_entries: list[tuple[str, _NodeEntry]] = []
for eid in stale_ids: for eid in stale_ids:
old = server.registry.pop(eid) old = server.registry.pop(eid)
stale_entries.append((eid, old)) stale_entries.append((eid, old))
server.registry[node_id] = entry server.registry[node_id] = entry
if entry.managed_assignment: if entry.managed_assignment:
if entry.hf_repo: if entry.hf_repo:
_rebalance_hf_model_locked(server, entry.hf_repo) _rebalance_hf_model_locked(server, entry.hf_repo)
else: else:
_rebalance_model_locked(server, model) _rebalance_model_locked(server, model)
assignment_directive = entry.pending_directives[-1] if entry.pending_directives else None assignment_directive = entry.pending_directives[-1] if entry.pending_directives else None
if assignment_directive is not None: if assignment_directive is not None:
entry.pending_directives.clear() entry.pending_directives.clear()
model_health = _model_health_summary(server, entry.model, entry.hf_repo) model_health = _model_health_summary(server, entry.model, entry.hf_repo)
for eid, old in stale_entries: for eid, old in stale_entries:
_tracker_log( _tracker_log(
server, server,
"info", "info",
"node re-registered", "node re-registered",
old_node_id=eid, old_node_id=eid,
node_id=node_id, node_id=node_id,
endpoint=old.endpoint, endpoint=old.endpoint,
old_model=old.model, old_model=old.model,
old_hf_repo=old.hf_repo, old_hf_repo=old.hf_repo,
old_model_health=_model_health_summary(server, old.model, old.hf_repo), old_model_health=_model_health_summary(server, old.model, old.hf_repo),
model_health=model_health, model_health=model_health,
) )
_tracker_log( _tracker_log(
server, server,
"info", "info",
"node registered", "node registered",
node_id=node_id, node_id=node_id,
endpoint=entry.endpoint, endpoint=entry.endpoint,
model=entry.model, model=entry.model,
hf_repo=entry.hf_repo, hf_repo=entry.hf_repo,
shard=f"{entry.shard_start}-{entry.shard_end}", shard=f"{entry.shard_start}-{entry.shard_end}",
tracker_mode=entry.tracker_mode, tracker_mode=entry.tracker_mode,
model_health=model_health, model_health=model_health,
) )
shard_info = f"layers {shard_start}-{shard_end}" if shard_start is not None else "unsharded" shard_info = f"layers {shard_start}-{shard_end}" if shard_start is not None else "unsharded"
repo_info = f" [{hf_repo}]" if hf_repo else "" repo_info = f" [{hf_repo}]" if hf_repo else ""
@@ -3312,6 +3352,8 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
entry.failed_requests = int(body["failed_requests"]) entry.failed_requests = int(body["failed_requests"])
if "queue_depth" in body: if "queue_depth" in body:
entry.queue_depth = int(body["queue_depth"]) entry.queue_depth = int(body["queue_depth"])
if "current_requests" in body:
entry.current_requests = _normalize_current_requests(body["current_requests"])
if "uptime_seconds" in body: if "uptime_seconds" in body:
entry.uptime_seconds = float(body["uptime_seconds"]) entry.uptime_seconds = float(body["uptime_seconds"])
if "status" in body and body["status"] in ("ready", "loading"): if "status" in body and body["status"] in ("ready", "loading"):
@@ -3593,7 +3635,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
return return
keys = server.accounts.keys_for(account["account_id"]) # type: ignore[union-attr] keys = server.accounts.keys_for(account["account_id"]) # type: ignore[union-attr]
balances = {} balances = {}
usage: dict = {"requests": 0, "total_tokens": 0, "total_cost": 0.0, "recent": []} usage: dict = {"requests": 0, "total_tokens": 0, "total_cost": 0.0, "recent": [], "records": []}
if server.billing is not None: if server.billing is not None:
balances = {key: server.billing.get_client_balance(key) for key in keys} balances = {key: server.billing.get_client_balance(key) for key in keys}
usage = server.billing.usage_for(keys) usage = server.billing.usage_for(keys)

4
tests/test_dashboard.py
View File

@@ -12,7 +12,9 @@ from meshnet_tracker.server import TrackerServer
PANELS = [ PANELS = [
"Tracker hive", "Nodes &amp; coverage", "Client balances", "Tracker hive", "Nodes &amp; coverage", "Client balances",
"Node pending payouts", "Settlement history", "Node pending payouts", "Settlement history",
"Strikes / bans / forfeitures", "Model usage", "Node throughput", "Strikes / bans / forfeitures", "Model usage", "Call wall",
"Usage summary", "Node throughput", "Request history",
"Chat / inference",
"Console output", "Console output",
] ]

72
tests/test_node_startup.py
View File

@@ -566,6 +566,78 @@ def test_download_shard_prefers_tracker_model_source_over_huggingface(
assert hf_calls == [] assert hf_calls == []
def test_download_shard_prefers_tracker_full_model_source_over_huggingface(
tmp_path,
monkeypatch,
):
"""A tracker-advertised full snapshot is sufficient on its own — HF is never contacted."""
contents = {
"config.json": b"{}",
"weights-a.safetensors": b"tracker-a",
"weights-b.safetensors": b"tracker-b",
}
class FakeFileResponse:
def __init__(self, payload: bytes):
self._payload = io.BytesIO(payload)
self._length = len(payload)
def __enter__(self):
return self
def __exit__(self, exc_type, exc, tb):
return False
def getheader(self, name: str):
if name == "Content-Length":
return str(self._length)
if name == "Content-Type":
return "application/octet-stream"
return None
def read(self, size: int = -1) -> bytes:
return self._payload.read(size)
def fake_urlopen(url, *args, **kwargs):
query = urllib.parse.parse_qs(urllib.parse.urlparse(url).query)
rel = query.get("file", [None])[0]
assert rel in contents, f"unexpected per-file request: {url}"
return FakeFileResponse(contents[rel])
monkeypatch.setattr(urllib.request, "urlopen", fake_urlopen)
hf_calls = []
def fake_snapshot_download(**kwargs):
hf_calls.append(kwargs)
raise AssertionError("HuggingFace should not be contacted when tracker full_files are available")
monkeypatch.setitem(
sys.modules,
"huggingface_hub",
types.SimpleNamespace(snapshot_download=fake_snapshot_download),
)
shard_dir = download_shard(
"tiny-llama",
0,
3,
cache_dir=tmp_path / "cache",
hf_repo="org/tiny-llama-shards",
model_sources=[{
"type": "tracker-full",
"url": "http://tracker/v1/model-files/download?model=tiny-llama&full=1",
"files": ["config.json", "weights-a.safetensors", "weights-b.safetensors"],
"full_files": ["config.json", "weights-a.safetensors", "weights-b.safetensors"],
}],
progress=False,
)
assert (shard_dir / "config.json").read_text() == "{}"
assert (shard_dir / "weights-a.safetensors").read_text() == "tracker-a"
assert (shard_dir / "weights-b.safetensors").read_text() == "tracker-b"
assert hf_calls == []
def test_download_shard_falls_back_to_huggingface_when_tracker_source_fails( def test_download_shard_falls_back_to_huggingface_when_tracker_source_fails(
tmp_path, tmp_path,
monkeypatch, monkeypatch,

293
tests/test_real_model_backend.py
View File

@@ -13,8 +13,12 @@ import pytest
from meshnet_node.model_backend import ( from meshnet_node.model_backend import (
InsufficientVRAMError, InsufficientVRAMError,
PartialModelLoadUnsupported,
TensorPayload, TensorPayload,
TorchModelShard,
_call_layer, _call_layer,
_load_partial_model_from_snapshot,
_should_partial_materialize_shard,
_decoder_attention_mask, _decoder_attention_mask,
_int_tensor_header, _int_tensor_header,
build_quantization_config, build_quantization_config,
@@ -399,6 +403,295 @@ def test_call_layer_passes_rotary_position_embeddings():
) == "hidden" ) == "hidden"
def test_partial_materialize_guard_requires_local_non_full_non_quantized_snapshot(tmp_path):
snapshot_dir = tmp_path / "snapshot"
snapshot_dir.mkdir()
(snapshot_dir / "config.json").write_text("{}")
(snapshot_dir / "model.safetensors.index.json").write_text('{"weight_map": {}}')
assert _should_partial_materialize_shard(
str(snapshot_dir),
4,
7,
total_layers_hint=40,
uses_quantized_weights=False,
) is True
assert _should_partial_materialize_shard(
str(snapshot_dir),
0,
39,
total_layers_hint=40,
uses_quantized_weights=False,
) is False
assert _should_partial_materialize_shard(
str(snapshot_dir),
4,
7,
total_layers_hint=40,
uses_quantized_weights=True,
) is False
assert _should_partial_materialize_shard(
"repo/model",
4,
7,
total_layers_hint=40,
uses_quantized_weights=False,
) is False
def test_partial_snapshot_loader_materializes_only_assigned_tensors(tmp_path):
snapshot_dir = tmp_path / "snapshot"
snapshot_dir.mkdir()
(snapshot_dir / "config.json").write_text("{}")
(snapshot_dir / "model.safetensors.index.json").write_text(json.dumps({
"weight_map": {
"model.embed_tokens.weight": "shard-1.safetensors",
"model.layers.0.self_attn.q_proj.weight": "shard-1.safetensors",
"model.layers.1.self_attn.q_proj.weight": "shard-2.safetensors",
"model.layers.2.self_attn.q_proj.weight": "shard-3.safetensors",
"model.norm.weight": "shard-3.safetensors",
"lm_head.weight": "shard-3.safetensors",
}
}))
for rel in ("shard-1.safetensors", "shard-2.safetensors", "shard-3.safetensors"):
(snapshot_dir / rel).write_bytes(b"stub")
class FakeModule:
def __init__(self, name):
self.name = name
self.to_calls = []
def to(self, device):
self.to_calls.append(device)
return self
class FakeModel:
def __init__(self):
self.model = types.SimpleNamespace(
embed_tokens=FakeModule("embed"),
layers=[FakeModule("layer0"), FakeModule("layer1"), FakeModule("layer2")],
rotary_emb=FakeModule("rotary"),
norm=FakeModule("norm"),
)
self.lm_head = FakeModule("lm_head")
self.tie_weights_called = 0
def tie_weights(self):
self.tie_weights_called += 1
class AutoConfigStub:
@staticmethod
def from_pretrained(model_id):
assert model_id == str(snapshot_dir)
return types.SimpleNamespace(num_hidden_layers=3)
class AutoModelStub:
@staticmethod
def from_config(cfg, torch_dtype=None):
assert cfg.num_hidden_layers == 3
assert torch_dtype == "bf16"
return FakeModel()
class EmptyWeights:
def __init__(self):
self.entered = 0
self.exited = 0
def __call__(self):
return self
def __enter__(self):
self.entered += 1
return None
def __exit__(self, exc_type, exc, tb):
self.exited += 1
return False
init_empty_weights = EmptyWeights()
set_calls = []
def fake_set_tensor(module, tensor_name, device, value=None, dtype=None):
set_calls.append((tensor_name, device, value, dtype))
tensors = {
"shard-1.safetensors": {
"model.embed_tokens.weight": "embed",
"model.layers.0.self_attn.q_proj.weight": "layer0",
},
"shard-2.safetensors": {
"model.layers.1.self_attn.q_proj.weight": "layer1",
},
"shard-3.safetensors": {
"model.layers.2.self_attn.q_proj.weight": "layer2",
"model.norm.weight": "norm",
"lm_head.weight": "lm_head",
},
}
class FakeSafeOpen:
def __init__(self, filename, framework, device):
assert framework == "pt"
assert device == "cpu"
self.filename = Path(filename).name
def __enter__(self):
return self
def __exit__(self, exc_type, exc, tb):
return False
def get_tensor(self, tensor_name):
return tensors[self.filename][tensor_name]
model = _load_partial_model_from_snapshot(
AutoConfigStub,
AutoModelStub,
types.SimpleNamespace(),
str(snapshot_dir),
1,
1,
"bf16",
"cpu:0",
init_empty_weights_fn=init_empty_weights,
set_tensor_fn=fake_set_tensor,
safe_open_fn=FakeSafeOpen,
)
assert init_empty_weights.entered == 1
assert init_empty_weights.exited == 1
assert model.tie_weights_called == 1
assert [call[0] for call in set_calls] == ["model.layers.1.self_attn.q_proj.weight"]
assert model.model.layers[1].to_calls == ["cpu:0"]
assert model.model.layers[0].to_calls == []
assert model.model.layers[2].to_calls == []
assert model.model.embed_tokens.to_calls == []
assert model.model.norm.to_calls == []
assert model.lm_head.to_calls == []
assert model.model.rotary_emb.to_calls == ["cpu:0"]
def test_partial_snapshot_loader_requires_known_layer_count(tmp_path):
snapshot_dir = tmp_path / "snapshot"
snapshot_dir.mkdir()
(snapshot_dir / "config.json").write_text("{}")
(snapshot_dir / "model.safetensors.index.json").write_text(json.dumps({
"weight_map": {"model.layers.0.self_attn.q_proj.weight": "shard.safetensors"}
}))
(snapshot_dir / "shard.safetensors").write_bytes(b"stub")
class AutoConfigStub:
@staticmethod
def from_pretrained(model_id):
return types.SimpleNamespace()
class AutoModelStub:
@staticmethod
def from_config(cfg, torch_dtype=None):
raise AssertionError("from_config should not run without a known layer count")
class UnusedContext:
def __enter__(self):
return None
def __exit__(self, exc_type, exc, tb):
return False
with pytest.raises(PartialModelLoadUnsupported, match="num_hidden_layers"):
_load_partial_model_from_snapshot(
AutoConfigStub,
AutoModelStub,
types.SimpleNamespace(),
str(snapshot_dir),
0,
0,
"bf16",
"cpu:0",
init_empty_weights_fn=lambda: UnusedContext(),
set_tensor_fn=lambda *args, **kwargs: None,
safe_open_fn=lambda *args, **kwargs: None,
)
def test_torch_model_shard_prefers_partial_loader_for_local_snapshot(tmp_path, monkeypatch):
import meshnet_node.model_backend as backend
snapshot_dir = tmp_path / "snapshot"
snapshot_dir.mkdir()
(snapshot_dir / "config.json").write_text("{}")
(snapshot_dir / "model.safetensors.index.json").write_text('{"weight_map": {}}')
class FakeModel:
def __init__(self):
self.model = types.SimpleNamespace(
layers=[object(), object(), object()],
embed_tokens=object(),
)
self.config = types.SimpleNamespace(hidden_size=8)
self.eval_called = 0
def eval(self):
self.eval_called += 1
fake_model = FakeModel()
partial_calls = []
class AutoConfigStub:
@staticmethod
def from_pretrained(model_id, cache_dir=None):
return types.SimpleNamespace(num_hidden_layers=3, text_config=types.SimpleNamespace(dtype="torch.bfloat16"))
class AutoModelStub:
@staticmethod
def from_pretrained(*args, **kwargs):
raise AssertionError("full model load should not run for partial local shards")
class AutoTokenizerStub:
@staticmethod
def from_pretrained(model_id, cache_dir=None):
assert model_id == str(snapshot_dir)
return types.SimpleNamespace()
monkeypatch.setitem(
sys.modules,
"torch",
types.SimpleNamespace(
cuda=types.SimpleNamespace(is_available=lambda: False),
device=lambda value: value,
bfloat16="bf16",
),
)
monkeypatch.setitem(
sys.modules,
"transformers",
types.SimpleNamespace(
AutoConfig=AutoConfigStub,
AutoModelForCausalLM=AutoModelStub,
AutoTokenizer=AutoTokenizerStub,
),
)
monkeypatch.setattr(
backend,
"_load_partial_model_from_snapshot",
lambda *args, **kwargs: partial_calls.append((args, kwargs)) or fake_model,
)
shard = TorchModelShard(
"repo/model",
1,
1,
quantization="auto",
cache_dir=snapshot_dir,
)
assert len(partial_calls) == 1
assert shard.model is fake_model
assert fake_model.eval_called == 1
assert shard.total_layers == 3
assert shard.is_head is False
assert shard.is_tail is False
@pytest.mark.integration @pytest.mark.integration
def test_two_node_gpt2_completion_is_deterministic(): def test_two_node_gpt2_completion_is_deterministic():
if os.environ.get("CI"): if os.environ.get("CI"):