11 Commits

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

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

4
.gitignore vendored
View File

@@ -20,3 +20,7 @@ dist/
.rocm-local/* .rocm-local/*
billing.sqlite billing.sqlite
.pytest-tmp/* .pytest-tmp/*
billing.sqlite
meshnet_registry.sqlite3
billing.sqlite
meshnet_registry.sqlite3

Binary file not shown.

Binary file not shown.

View File

@@ -134,8 +134,9 @@ class TorchModelShard:
self.model.to(self.device) self.model.to(self.device)
except Exception as exc: except Exception as exc:
if _looks_like_oom(exc): if _looks_like_oom(exc):
memory_kind = "VRAM" if self.device.type == "cuda" else "RAM"
raise InsufficientVRAMError( raise InsufficientVRAMError(
f"insufficient VRAM to load {model_id} layers {shard_start}:{shard_end} " f"insufficient {memory_kind} to load {model_id} layers {shard_start}:{shard_end} "
f"with {quantization} quantization; choose a smaller shard or lower quantization" f"with {quantization} quantization; choose a smaller shard or lower quantization"
) from exc ) from exc
raise raise
@@ -215,7 +216,7 @@ class TorchModelShard:
def generate_text( def generate_text(
self, self,
messages: list[dict], messages: list[dict],
max_new_tokens: int = 256, max_new_tokens: int = 5120,
temperature: float = 1.0, temperature: float = 1.0,
top_p: float = 1.0, top_p: float = 1.0,
) -> str: ) -> str:
@@ -245,7 +246,7 @@ class TorchModelShard:
def generate_text_streaming( def generate_text_streaming(
self, self,
messages: list[dict], messages: list[dict],
max_new_tokens: int = 256, max_new_tokens: int = 5000,
temperature: float = 1.0, temperature: float = 1.0,
top_p: float = 1.0, top_p: float = 1.0,
): ):
@@ -411,7 +412,7 @@ def _should_partial_materialize_shard(
return False return False
if total_layers_hint is None: if total_layers_hint is None:
return False return False
return not (shard_start == 0 and shard_end >= total_layers_hint - 1) return True
def _load_partial_model_from_snapshot( def _load_partial_model_from_snapshot(
@@ -476,7 +477,7 @@ def _load_partial_model_from_snapshot(
) )
with init_empty_weights_fn(): with init_empty_weights_fn():
model = auto_model_for_causal_lm.from_config(cfg, torch_dtype=dtype) model = auto_model_for_causal_lm.from_config(_causal_lm_config(cfg), torch_dtype=dtype)
tie_weights = getattr(model, "tie_weights", None) tie_weights = getattr(model, "tie_weights", None)
if callable(tie_weights): if callable(tie_weights):
tie_weights() tie_weights()
@@ -498,7 +499,7 @@ def _load_partial_model_from_snapshot(
for tensor_name in names: for tensor_name in names:
set_tensor_fn( set_tensor_fn(
model, model,
tensor_name, _checkpoint_tensor_name_for_model(model, tensor_name),
device, device,
value=handle.get_tensor(tensor_name), value=handle.get_tensor(tensor_name),
dtype=dtype, dtype=dtype,
@@ -569,38 +570,74 @@ def _native_torch_dtype(cfg: Any, torch: Any) -> Any:
return torch.bfloat16 return torch.bfloat16
def _causal_lm_config(cfg: Any) -> Any:
"""Use the text decoder config for composite VLM/MoE presets."""
get_text_config = getattr(cfg, "get_text_config", None)
if callable(get_text_config):
try:
return get_text_config()
except Exception:
pass
text_config = getattr(cfg, "text_config", None)
if text_config is not None:
return text_config
return cfg
def _checkpoint_tensor_name_for_model(model: Any, tensor_name: str) -> str:
"""Map multimodal checkpoint keys onto text-only CausalLM modules when needed."""
inner = getattr(model, "model", None)
if inner is not None and hasattr(inner, "language_model"):
return tensor_name
if ".language_model." in tensor_name:
return tensor_name.replace(".language_model.", ".")
return tensor_name
def _transformer_backbone(model: Any) -> Any:
if hasattr(model, "model"):
inner = model.model
language_model = getattr(inner, "language_model", None)
if language_model is not None:
return language_model
return inner
if hasattr(model, "transformer"):
return model.transformer
raise ModelBackendError(
"unsupported HuggingFace model architecture: no transformer backbone found"
)
def _model_layers(model: Any) -> Any: def _model_layers(model: Any) -> Any:
if hasattr(model, "model") and hasattr(model.model, "layers"): backbone = _transformer_backbone(model)
return model.model.layers for attr in ("layers", "h", "blocks"):
if hasattr(model, "transformer") and hasattr(model.transformer, "h"): layers = getattr(backbone, attr, None)
return model.transformer.h if layers is not None:
return layers
raise ModelBackendError( raise ModelBackendError(
"unsupported HuggingFace model architecture: no transformer layers found" "unsupported HuggingFace model architecture: no transformer layers found"
) )
def _embed_tokens(model: Any) -> Any: def _embed_tokens(model: Any) -> Any:
if hasattr(model, "model") and hasattr(model.model, "embed_tokens"): backbone = _transformer_backbone(model)
return model.model.embed_tokens for attr in ("embed_tokens", "wte"):
if hasattr(model, "transformer") and hasattr(model.transformer, "wte"): embed = getattr(backbone, attr, None)
return model.transformer.wte if embed is not None:
return embed
raise ModelBackendError( raise ModelBackendError(
"unsupported HuggingFace model architecture: no token embeddings found" "unsupported HuggingFace model architecture: no token embeddings found"
) )
def _position_embeddings(model: Any) -> Any | None: def _position_embeddings(model: Any) -> Any | None:
if hasattr(model, "transformer") and hasattr(model.transformer, "wpe"): backbone = _transformer_backbone(model)
return model.transformer.wpe return getattr(backbone, "wpe", None)
return None
def _rotary_embedding_module(model: Any) -> Any | None: def _rotary_embedding_module(model: Any) -> Any | None:
if hasattr(model, "model") and hasattr(model.model, "rotary_emb"): backbone = _transformer_backbone(model)
return model.model.rotary_emb return getattr(backbone, "rotary_emb", None)
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]: def _active_modules_for_shard(model: Any, shard_start: int, shard_end: int) -> list[Any]:
@@ -627,10 +664,11 @@ def _active_modules_for_shard(model: Any, shard_start: int, shard_end: int) -> l
def _final_norm(model: Any) -> Any | None: def _final_norm(model: Any) -> Any | None:
if hasattr(model, "model") and hasattr(model.model, "norm"): backbone = _transformer_backbone(model)
return model.model.norm for attr in ("norm", "ln_f", "final_layer_norm"):
if hasattr(model, "transformer") and hasattr(model.transformer, "ln_f"): norm = getattr(backbone, attr, None)
return model.transformer.ln_f if norm is not None:
return norm
return None return None
@@ -743,7 +781,12 @@ def _looks_like_oom(exc: BaseException) -> bool:
current: BaseException | None = exc current: BaseException | None = exc
while current is not None: while current is not None:
text = str(current).lower() text = str(current).lower()
if "out of memory" in text or "cuda error: out of memory" in text: if (
"out of memory" in text
or "cuda error: out of memory" in text
or "paging file is too small" in text
or "os error 1455" in text
):
return True return True
current = current.__cause__ or current.__context__ current = current.__cause__ or current.__context__
return False return False

View File

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

View File

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

View File

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

View File

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

View File

@@ -35,6 +35,7 @@ import urllib.parse
import urllib.request import urllib.request
import uuid import uuid
from collections import deque from collections import deque
from dataclasses import dataclass, field
from importlib.resources import files from importlib.resources import files
from pathlib import Path from pathlib import Path
from typing import Any from typing import Any
@@ -51,6 +52,7 @@ from .raft import RaftNode
_CONSOLE_LIMIT = 300 _CONSOLE_LIMIT = 300
_PROXY_PROGRESS_LOG_INTERVAL = 5.0
def _preset_price_keys(name: str, preset: dict) -> set[str]: def _preset_price_keys(name: str, preset: dict) -> set[str]:
@@ -1413,6 +1415,10 @@ def _relay_http_request_frames(
headers: dict[str, str], headers: dict[str, str],
timeout: float = 310.0, timeout: float = 310.0,
idle_timeout: float = 120.0, idle_timeout: float = 120.0,
*,
cancel_event: threading.Event | None = None,
ws_holder: list[Any] | None = None,
ws_lock: threading.Lock | None = None,
): ):
"""Send an HTTP-shaped request through a relay RPC WebSocket, yielding """Send an HTTP-shaped request through a relay RPC WebSocket, yielding
response frames until a terminal one (US-036). response frames until a terminal one (US-036).
@@ -1430,6 +1436,14 @@ def _relay_http_request_frames(
deadline = time.monotonic() + timeout deadline = time.monotonic() + timeout
try: try:
with wsc.connect(relay_addr, open_timeout=10, close_timeout=5) as ws: with wsc.connect(relay_addr, open_timeout=10, close_timeout=5) as ws:
if ws_holder is not None:
if ws_lock is not None:
with ws_lock:
ws_holder.clear()
ws_holder.append(ws)
else:
ws_holder.clear()
ws_holder.append(ws)
ws.send(json.dumps({ ws.send(json.dumps({
"request_id": request_id, "request_id": request_id,
"method": "POST", "method": "POST",
@@ -1438,6 +1452,8 @@ def _relay_http_request_frames(
"body": body.decode(errors="replace"), "body": body.decode(errors="replace"),
})) }))
while True: while True:
if cancel_event is not None and cancel_event.is_set():
return
remaining = deadline - time.monotonic() remaining = deadline - time.monotonic()
if remaining <= 0: if remaining <= 0:
return return
@@ -2076,7 +2092,15 @@ def _registration_ban_error(contracts: Any | None, wallet_address: str | None) -
return None return None
def _tracker_log(server: "_TrackerHTTPServer", level: str, message: str, **fields: Any) -> None: def _tracker_log(
server: "_TrackerHTTPServer",
level: str,
message: str,
*,
stdout: bool = True,
update_console_key: str | None = None,
**fields: Any,
) -> None:
event = { event = {
"ts": time.time(), "ts": time.time(),
"level": level, "level": level,
@@ -2088,10 +2112,88 @@ def _tracker_log(server: "_TrackerHTTPServer", level: str, message: str, **field
}, },
} }
with server.console_lock: with server.console_lock:
server.console_events.append(event) if update_console_key is not None:
extras = " ".join(f"{key}={value}" for key, value in event["fields"].items()) updated = False
suffix = f" {extras}" if extras else "" for existing in reversed(server.console_events):
print(f"[tracker] {level}: {message}{suffix}", flush=True) if (
existing.get("message") == message
and existing.get("fields", {}).get("request_id") == update_console_key
):
existing["ts"] = event["ts"]
existing["fields"] = event["fields"]
updated = True
break
if not updated:
server.console_events.append(event)
else:
server.console_events.append(event)
if stdout:
extras = " ".join(f"{key}={value}" for key, value in event["fields"].items())
suffix = f" {extras}" if extras else ""
print(f"[tracker] {level}: {message}{suffix}", flush=True)
@dataclass
class _ActiveProxyContext:
request_id: str
cancel_event: threading.Event = field(default_factory=threading.Event)
upstream: Any | None = None
upstream_lock: threading.Lock = field(default_factory=threading.Lock)
relay_ws: Any | None = None
relay_ws_lock: threading.Lock = field(default_factory=threading.Lock)
def _register_active_proxy(server: "_TrackerHTTPServer", request_id: str) -> _ActiveProxyContext:
ctx = _ActiveProxyContext(request_id=request_id)
with server.active_proxies_lock:
server.active_proxies[request_id] = ctx
return ctx
def _unregister_active_proxy(server: "_TrackerHTTPServer", request_id: str) -> None:
with server.active_proxies_lock:
server.active_proxies.pop(request_id, None)
def _request_proxy_cancel(server: "_TrackerHTTPServer", request_id: str) -> bool:
with server.active_proxies_lock:
ctx = server.active_proxies.get(request_id)
if ctx is None:
return False
ctx.cancel_event.set()
def _close_resources() -> None:
with ctx.upstream_lock:
upstream = ctx.upstream
if upstream is not None:
try:
upstream.close()
except Exception:
pass
with ctx.relay_ws_lock:
relay_ws = ctx.relay_ws
if relay_ws is not None:
try:
relay_ws.close()
except Exception:
pass
threading.Thread(target=_close_resources, daemon=True).start()
return True
def _set_upstream_read_timeout(upstream: Any, timeout: float) -> None:
fp = getattr(upstream, "fp", None)
raw = getattr(fp, "raw", None) if fp is not None else None
sock = getattr(raw, "_sock", None) if raw is not None else None
if sock is not None:
sock.settimeout(timeout)
def _clear_proxy_progress_log_state(server: "_TrackerHTTPServer", request_id: str) -> None:
state = getattr(server, "_proxy_progress_log_state", None)
if state is not None:
state.pop(request_id, None)
def _tracker_log_proxy_progress( def _tracker_log_proxy_progress(
@@ -2108,10 +2210,21 @@ def _tracker_log_proxy_progress(
) -> None: ) -> None:
elapsed = time.monotonic() - started elapsed = time.monotonic() - started
effective_elapsed = max(elapsed, 1e-6) effective_elapsed = max(elapsed, 1e-6)
now = time.monotonic()
state = getattr(server, "_proxy_progress_log_state", None)
if state is None:
state = {}
server._proxy_progress_log_state = state
last_stdout = state.get(request_id)
stdout = last_stdout is None or (now - last_stdout) >= _PROXY_PROGRESS_LOG_INTERVAL
if stdout:
state[request_id] = now
_tracker_log( _tracker_log(
server, server,
"info", "info",
"proxy progress", "proxy progress",
stdout=stdout,
update_console_key=request_id,
request_id=request_id, request_id=request_id,
model=model, model=model,
route_model=route_model, route_model=route_model,
@@ -2209,6 +2322,8 @@ class _TrackerHTTPServer(socketserver.ThreadingMixIn, http.server.HTTPServer):
self.models_dir = models_dir self.models_dir = models_dir
self.console_events = deque(maxlen=_CONSOLE_LIMIT) self.console_events = deque(maxlen=_CONSOLE_LIMIT)
self.console_lock = threading.Lock() self.console_lock = threading.Lock()
self.active_proxies: dict[str, _ActiveProxyContext] = {}
self.active_proxies_lock = threading.Lock()
class _TrackerHandler(http.server.BaseHTTPRequestHandler): class _TrackerHandler(http.server.BaseHTTPRequestHandler):
@@ -2364,6 +2479,17 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
if len(parts) == 5 and parts[1] == "v1" and parts[2] == "nodes" and parts[4] == "heartbeat": if len(parts) == 5 and parts[1] == "v1" and parts[2] == "nodes" and parts[4] == "heartbeat":
self._handle_heartbeat(parts[3]) self._handle_heartbeat(parts[3])
return return
# /v1/proxy/requests/<request_id>/cancel
if (
len(parts) == 6
and parts[1] == "v1"
and parts[2] == "proxy"
and parts[3] == "requests"
and parts[5] == "cancel"
and parts[4]
):
self._handle_proxy_request_cancel(urllib.parse.unquote(parts[4]))
return
self.send_response(404) self.send_response(404)
self.end_headers() self.end_headers()
@@ -2886,6 +3012,9 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
if inflight_recorded: if inflight_recorded:
_record_proxy_inflight(server, inflight_nodes, -1) _record_proxy_inflight(server, inflight_nodes, -1)
inflight_recorded = False inflight_recorded = False
_unregister_active_proxy(server, request_id)
proxy_ctx = _register_active_proxy(server, request_id)
_tracker_log( _tracker_log(
server, server,
@@ -2906,6 +3035,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
headers={ headers={
"Content-Type": "application/json", "Content-Type": "application/json",
"X-Meshnet-Route": downstream_urls, "X-Meshnet-Route": downstream_urls,
"X-Meshnet-Request-Id": request_id,
}, },
method="POST", method="POST",
) )
@@ -2917,6 +3047,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
relay_headers = { relay_headers = {
"Content-Type": "application/json", "Content-Type": "application/json",
"X-Meshnet-Route": downstream_urls, "X-Meshnet-Route": downstream_urls,
"X-Meshnet-Request-Id": request_id,
**({"Authorization": auth} if auth else {}), **({"Authorization": auth} if auth else {}),
} }
@@ -2930,13 +3061,34 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
direct_endpoint=target_url, direct_endpoint=target_url,
) )
started = time.monotonic() started = time.monotonic()
relay_ws_holder: list[Any] = []
frames = _relay_http_request_frames( frames = _relay_http_request_frames(
node.relay_addr, node.relay_addr,
path="/v1/chat/completions", path="/v1/chat/completions",
body=raw_body, body=raw_body,
headers=relay_headers, headers=relay_headers,
cancel_event=proxy_ctx.cancel_event,
ws_holder=relay_ws_holder,
ws_lock=proxy_ctx.relay_ws_lock,
) )
first = next(frames, None) first = next(frames, None)
with proxy_ctx.relay_ws_lock:
proxy_ctx.relay_ws = relay_ws_holder[0] if relay_ws_holder else None
if proxy_ctx.cancel_event.is_set():
if self._finalize_proxy_cancel(
proxy_ctx=proxy_ctx,
server=server,
request_id=request_id,
started=started,
model=model,
route_model=route_model,
route_nodes=route_nodes,
api_key=api_key,
node_work=node_work,
body=body,
finish_proxy_inflight=finish_proxy_inflight,
):
return
if first is not None and first.get("stream"): if first is not None and first.get("stream"):
# Streamed response (US-036): forward SSE chunks as they arrive # Streamed response (US-036): forward SSE chunks as they arrive
# and run the same token accounting as the direct stream path. # and run the same token accounting as the direct stream path.
@@ -2945,6 +3097,8 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
model, route_model, route_nodes, api_key, node_work, model, route_model, route_nodes, api_key, node_work,
request_body=body, request_body=body,
request_id=request_id, request_id=request_id,
proxy_ctx=proxy_ctx,
finish_proxy_inflight=finish_proxy_inflight,
) )
finish_proxy_inflight() finish_proxy_inflight()
return return
@@ -2959,6 +3113,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
in_tokens, out_tokens = 0, 0 in_tokens, out_tokens = 0, 0
tokens = in_tokens + out_tokens tokens = in_tokens + out_tokens
self._record_observed_throughput(model, route_model, tokens, elapsed, route_nodes) self._record_observed_throughput(model, route_model, tokens, elapsed, route_nodes)
_clear_proxy_progress_log_state(server, request_id)
_tracker_log( _tracker_log(
server, server,
"info", "info",
@@ -2989,7 +3144,69 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
try: try:
started = time.monotonic() started = time.monotonic()
upstream = urllib.request.urlopen(req, timeout=300.0) upstream_result: list[Any] = []
connect_errors: list[BaseException] = []
def _connect_upstream() -> None:
try:
upstream_result.append(urllib.request.urlopen(req, timeout=300.0))
except BaseException as exc:
connect_errors.append(exc)
connect_thread = threading.Thread(target=_connect_upstream, daemon=True)
connect_thread.start()
while connect_thread.is_alive():
if proxy_ctx.cancel_event.is_set():
connect_thread.join(timeout=310.0)
if upstream_result:
try:
upstream_result[0].close()
except Exception:
pass
if self._finalize_proxy_cancel(
proxy_ctx=proxy_ctx,
server=server,
request_id=request_id,
started=started,
model=model,
route_model=route_model,
route_nodes=route_nodes,
api_key=api_key,
node_work=node_work,
body=body,
finish_proxy_inflight=finish_proxy_inflight,
):
return
connect_thread.join(0.2)
if proxy_ctx.cancel_event.is_set():
if upstream_result:
try:
upstream_result[0].close()
except Exception:
pass
if self._finalize_proxy_cancel(
proxy_ctx=proxy_ctx,
server=server,
request_id=request_id,
started=started,
model=model,
route_model=route_model,
route_nodes=route_nodes,
api_key=api_key,
node_work=node_work,
body=body,
finish_proxy_inflight=finish_proxy_inflight,
):
return
if connect_errors:
raise connect_errors[0]
upstream = upstream_result[0]
with proxy_ctx.upstream_lock:
proxy_ctx.upstream = upstream
_set_upstream_read_timeout(upstream, 0.5)
_tracker_log(server, "info", "proxy connected", request_id=request_id, target_url=target_url) _tracker_log(server, "info", "proxy connected", request_id=request_id, target_url=target_url)
except urllib.error.HTTPError as exc: except urllib.error.HTTPError as exc:
# Relay error status + body from node # Relay error status + body from node
@@ -3002,9 +3219,11 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
self.wfile.write(err_body) self.wfile.write(err_body)
except BrokenPipeError: except BrokenPipeError:
pass pass
_clear_proxy_progress_log_state(server, request_id)
finish_proxy_inflight() finish_proxy_inflight()
return return
except Exception as exc: except Exception as exc:
_clear_proxy_progress_log_state(server, request_id)
if node.relay_addr: if node.relay_addr:
_tracker_log( _tracker_log(
server, server,
@@ -3045,8 +3264,15 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
client_gone = False client_gone = False
try: try:
while True: while True:
line = upstream.readline() if proxy_ctx.cancel_event.is_set():
break
try:
line = upstream.readline()
except TimeoutError:
continue
if not line: if not line:
if proxy_ctx.cancel_event.is_set():
break
break break
if not client_gone: if not client_gone:
try: try:
@@ -3071,6 +3297,22 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
stream_usage = usage stream_usage = usage
except (BrokenPipeError, ConnectionResetError): except (BrokenPipeError, ConnectionResetError):
pass pass
if self._finalize_proxy_cancel(
proxy_ctx=proxy_ctx,
server=server,
request_id=request_id,
started=started,
model=model,
route_model=route_model,
route_nodes=route_nodes,
api_key=api_key,
node_work=node_work,
body=body,
finish_proxy_inflight=finish_proxy_inflight,
observed_stream_tokens=observed_stream_tokens,
stream_usage=stream_usage,
):
return
elapsed = time.monotonic() - started elapsed = time.monotonic() - started
# Bill even on client disconnect — the nodes did the work. # Bill even on client disconnect — the nodes did the work.
# Observed stream chunks are authoritative for the upper bound; # Observed stream chunks are authoritative for the upper bound;
@@ -3082,6 +3324,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
model, route_model, in_tokens + out_tokens, elapsed, route_nodes model, route_model, in_tokens + out_tokens, elapsed, route_nodes
) )
tokens = in_tokens + out_tokens tokens = in_tokens + out_tokens
_clear_proxy_progress_log_state(server, request_id)
_tracker_log( _tracker_log(
server, server,
"info", "info",
@@ -3113,6 +3356,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
in_tokens, out_tokens = 0, 0 in_tokens, out_tokens = 0, 0
tokens = in_tokens + out_tokens tokens = in_tokens + out_tokens
self._record_observed_throughput(model, route_model, tokens, elapsed, route_nodes) self._record_observed_throughput(model, route_model, tokens, elapsed, route_nodes)
_clear_proxy_progress_log_state(server, request_id)
_tracker_log( _tracker_log(
server, server,
"info", "info",
@@ -3272,6 +3516,9 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
node_work: list, node_work: list,
request_body: dict, request_body: dict,
request_id: str, request_id: str,
*,
proxy_ctx: _ActiveProxyContext | None = None,
finish_proxy_inflight: Any = None,
) -> None: ) -> None:
"""Forward a streamed relay response (US-036) to the client as SSE, """Forward a streamed relay response (US-036) to the client as SSE,
billing with the same accounting as the direct stream path.""" billing with the same accounting as the direct stream path."""
@@ -3285,6 +3532,8 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
observed_stream_tokens = 0 observed_stream_tokens = 0
client_gone = False client_gone = False
for frame in itertools.chain([first], frames): for frame in itertools.chain([first], frames):
if proxy_ctx is not None and proxy_ctx.cancel_event.is_set():
break
chunk = frame.get("chunk") or "" chunk = frame.get("chunk") or ""
if not chunk: if not chunk:
continue continue
@@ -3312,6 +3561,26 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
) )
if usage is not None: if usage is not None:
stream_usage = usage stream_usage = usage
if (
proxy_ctx is not None
and finish_proxy_inflight is not None
and self._finalize_proxy_cancel(
proxy_ctx=proxy_ctx,
server=server,
request_id=request_id,
started=started,
model=model,
route_model=route_model,
route_nodes=route_nodes,
api_key=api_key,
node_work=node_work,
body=request_body,
finish_proxy_inflight=finish_proxy_inflight,
observed_stream_tokens=observed_stream_tokens,
stream_usage=stream_usage,
)
):
return
elapsed = time.monotonic() - started elapsed = time.monotonic() - started
in_tokens, out_tokens = _stream_billable_split( in_tokens, out_tokens = _stream_billable_split(
observed_stream_tokens, stream_usage, request_body observed_stream_tokens, stream_usage, request_body
@@ -3320,6 +3589,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
model, route_model, in_tokens + out_tokens, elapsed, route_nodes model, route_model, in_tokens + out_tokens, elapsed, route_nodes
) )
tokens = in_tokens + out_tokens tokens = in_tokens + out_tokens
_clear_proxy_progress_log_state(server, request_id)
_tracker_log( _tracker_log(
server, server,
"info", "info",
@@ -3765,6 +4035,68 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
events = [dict(event) for event in server.console_events] events = [dict(event) for event in server.console_events]
self._send_json(200, {"events": events}) self._send_json(200, {"events": events})
def _handle_proxy_request_cancel(self, request_id: str) -> None:
server: _TrackerHTTPServer = self.server # type: ignore[assignment]
if server.accounts is not None and not self._require_role("admin"):
return
if not _request_proxy_cancel(server, request_id):
self._send_json(404, {"error": f"no active proxy for request {request_id!r}"})
return
self._send_json(200, {"status": "canceled", "request_id": request_id})
def _finalize_proxy_cancel(
self,
*,
proxy_ctx: _ActiveProxyContext,
server: "_TrackerHTTPServer",
request_id: str,
started: float,
model: str,
route_model: str,
route_nodes: list,
api_key: str | None,
node_work: list,
body: dict,
finish_proxy_inflight: Any,
observed_stream_tokens: int = 0,
stream_usage: dict | None = None,
) -> bool:
if not proxy_ctx.cancel_event.is_set():
return False
elapsed = time.monotonic() - started
_clear_proxy_progress_log_state(server, request_id)
tokens = observed_stream_tokens
if observed_stream_tokens > 0:
in_tokens, out_tokens = _stream_billable_split(
observed_stream_tokens, stream_usage, body,
)
tokens = in_tokens + out_tokens
self._record_observed_throughput(
model, route_model, tokens, elapsed, route_nodes,
)
_tracker_log(
server,
"info",
"proxy canceled",
request_id=request_id,
model=model,
route_model=route_model,
tokens=tokens,
elapsed_seconds=round(elapsed, 4),
tokens_per_sec=round(tokens / elapsed, 4) if elapsed > 0 and tokens > 0 else 0.0,
route=_node_route_summary(route_nodes),
)
if observed_stream_tokens > 0:
in_tokens, out_tokens = _stream_billable_split(
observed_stream_tokens, stream_usage, body,
)
self._bill_completed(
api_key, model, in_tokens + out_tokens, node_work,
input_tokens=in_tokens, output_tokens=out_tokens,
)
finish_proxy_inflight()
return True
def _handle_registry_wallets(self): def _handle_registry_wallets(self):
server: _TrackerHTTPServer = self.server # type: ignore[assignment] server: _TrackerHTTPServer = self.server # type: ignore[assignment]
if not self._require_role("admin"): if not self._require_role("admin"):
@@ -4532,6 +4864,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
"model": resolved_name, "model": resolved_name,
"model_layers_end": required_end, "model_layers_end": required_end,
"peers": peers, "peers": peers,
"bytes_per_layer": _preset_bytes_per_layer(preset),
"model_sources": self._model_sources( "model_sources": self._model_sources(
resolved_name, resolved_name,
preset, preset,

View File

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

View File

@@ -17,6 +17,7 @@ from meshnet_node.model_backend import (
TensorPayload, TensorPayload,
TorchModelShard, TorchModelShard,
_call_layer, _call_layer,
_checkpoint_tensor_name_for_model,
_load_partial_model_from_snapshot, _load_partial_model_from_snapshot,
_should_partial_materialize_shard, _should_partial_materialize_shard,
_decoder_attention_mask, _decoder_attention_mask,
@@ -225,7 +226,7 @@ def test_tail_forward_returns_text_completion_from_binary_activations():
node.stop() node.stop()
def test_full_model_chat_completion_uses_generation_not_single_token_decode(): def test_full_model_chat_completion_uses_generation_not_single_token_decode(capsys):
node = TorchNodeServer(backend=_FakeFullBackend()) node = TorchNodeServer(backend=_FakeFullBackend())
port = node.start() port = node.start()
try: try:
@@ -237,7 +238,10 @@ def test_full_model_chat_completion_uses_generation_not_single_token_decode():
req = urllib.request.Request( req = urllib.request.Request(
f"http://127.0.0.1:{port}/v1/chat/completions", f"http://127.0.0.1:{port}/v1/chat/completions",
data=payload, data=payload,
headers={"Content-Type": "application/json"}, headers={
"Content-Type": "application/json",
"X-Meshnet-Request-Id": "req-test-123",
},
method="POST", method="POST",
) )
with urllib.request.urlopen(req, timeout=5) as resp: with urllib.request.urlopen(req, timeout=5) as resp:
@@ -248,6 +252,10 @@ def test_full_model_chat_completion_uses_generation_not_single_token_decode():
finally: finally:
node.stop() node.stop()
out = capsys.readouterr().out
assert " [node] processing chat model='fake-model' stream=False max_tokens=7 request_id=req-test-123" in out
assert " [node] chat complete tokens=1 elapsed_s=" in out
def test_pipeline_hop_logs_are_suppressed_without_debug(capsys): def test_pipeline_hop_logs_are_suppressed_without_debug(capsys):
tail_backend = _FakePipelineTailBackend() tail_backend = _FakePipelineTailBackend()
@@ -422,7 +430,7 @@ def test_partial_materialize_guard_requires_local_non_full_non_quantized_snapsho
39, 39,
total_layers_hint=40, total_layers_hint=40,
uses_quantized_weights=False, uses_quantized_weights=False,
) is False ) is True
assert _should_partial_materialize_shard( assert _should_partial_materialize_shard(
str(snapshot_dir), str(snapshot_dir),
4, 4,
@@ -439,6 +447,118 @@ def test_partial_materialize_guard_requires_local_non_full_non_quantized_snapsho
) is False ) is False
def test_checkpoint_tensor_name_remapped_for_text_only_causal_lm():
class TextOnlyModel:
def __init__(self):
self.model = types.SimpleNamespace(layers=[])
model = TextOnlyModel()
assert _checkpoint_tensor_name_for_model(
model,
"model.language_model.layers.0.mlp.gate.weight",
) == "model.layers.0.mlp.gate.weight"
assert _checkpoint_tensor_name_for_model(
model,
"model.language_model.embed_tokens.weight",
) == "model.embed_tokens.weight"
def test_checkpoint_tensor_name_kept_for_multimodal_backbone():
class MultimodalModel:
def __init__(self):
self.model = types.SimpleNamespace(language_model=types.SimpleNamespace())
model = MultimodalModel()
name = "model.language_model.layers.0.mlp.gate.weight"
assert _checkpoint_tensor_name_for_model(model, name) == name
def test_partial_snapshot_loader_remaps_language_model_checkpoint_keys(tmp_path):
snapshot_dir = tmp_path / "snapshot"
snapshot_dir.mkdir()
(snapshot_dir / "config.json").write_text(json.dumps({
"text_config": {"num_hidden_layers": 3},
}))
(snapshot_dir / "model.safetensors.index.json").write_text(json.dumps({
"weight_map": {
"model.language_model.layers.1.self_attn.q_proj.weight": "shard-2.safetensors",
}
}))
(snapshot_dir / "shard-2.safetensors").write_bytes(b"stub")
class FakeModule:
def __init__(self):
self.to_calls = []
def to(self, device):
self.to_calls.append(device)
return self
class FakeModel:
def __init__(self):
self.model = types.SimpleNamespace(
layers=[FakeModule(), FakeModule(), FakeModule()],
rotary_emb=FakeModule(),
)
def tie_weights(self):
pass
class AutoConfigStub:
@staticmethod
def from_pretrained(model_id):
return types.SimpleNamespace(
text_config=types.SimpleNamespace(num_hidden_layers=3),
get_text_config=lambda: types.SimpleNamespace(num_hidden_layers=3),
)
class AutoModelStub:
@staticmethod
def from_config(cfg, torch_dtype=None):
return FakeModel()
set_calls = []
def fake_set_tensor(module, tensor_name, device, value=None, dtype=None):
set_calls.append(tensor_name)
class FakeSafeOpen:
def __init__(self, filename, framework, device):
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 tensor_name
class UnusedContext:
def __enter__(self):
return None
def __exit__(self, exc_type, exc, tb):
return False
_load_partial_model_from_snapshot(
AutoConfigStub,
AutoModelStub,
types.SimpleNamespace(),
str(snapshot_dir),
1,
1,
"bf16",
"cpu:0",
init_empty_weights_fn=UnusedContext,
set_tensor_fn=fake_set_tensor,
safe_open_fn=FakeSafeOpen,
)
assert set_calls == ["model.layers.1.self_attn.q_proj.weight"]
def test_partial_snapshot_loader_materializes_only_assigned_tensors(tmp_path): def test_partial_snapshot_loader_materializes_only_assigned_tensors(tmp_path):
snapshot_dir = tmp_path / "snapshot" snapshot_dir = tmp_path / "snapshot"
snapshot_dir.mkdir() snapshot_dir.mkdir()

View File

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