misc

.gitignore

@@ -19,4 +19,3 @@ dist/
 !.env.testnet
 .rocm-local/*
 billing.sqlite
-.pytest-tmp/*

packages/node/meshnet_node/cli.py

@@ -90,19 +90,6 @@ def _run_node(cfg: dict) -> None:
         )
 
 
-def _resolve_model_flags(
-    model: str | None,
-    model_id: str | None,
-) -> tuple[str | None, str | None]:
-    """Return (model_name, hf_repo_or_none) from --model / --model-id flags."""
-    explicit = model_id or model
-    if not explicit:
-        return None, None
-    if "/" in explicit:
-        return explicit.split("/")[-1], explicit
-    return explicit, None
-
-
 def _first_available_port(host: str, start: int = 7000, attempts: int = 100) -> int:
     """Return the first TCP port bindable on host, starting at start."""
     bind_host = "" if host == "0.0.0.0" else host
@@ -135,10 +122,9 @@ def _cmd_default(args) -> int:
 
     # Apply CLI overrides on top of saved config
     overrides: dict = {}
-    model_name, hf_repo = _resolve_model_flags(args.model, getattr(args, "model_id", None))
-    if model_name is not None:
-        overrides["model_name"] = model_name
-        overrides["model_hf_repo"] = hf_repo or ""
+    if args.model:
+        overrides["model_hf_repo"] = args.model
+        overrides["model_name"] = args.model.split("/")[-1]
     if args.quantization:
         overrides["quantization"] = args.quantization
     if args.download_dir:
@@ -229,15 +215,16 @@ def _cmd_start(args) -> int:
     if args.tracker:
         cfg["tracker_url"] = args.tracker
     cfg["port"] = args.port if args.port is not None else _first_available_port(args.host)
-    model_name, hf_repo = _resolve_model_flags(
-        args.model or cfg.get("model_hf_repo") or cfg.get("model_name") or None,
-        args.model_id,
-    )
-    if model_name is not None:
-        cfg["model_name"] = model_name
-        cfg["model_hf_repo"] = hf_repo or ""
+    model = args.model or cfg.get("model_hf_repo") or cfg.get("model_name") or None
+    if args.model_id is None and model and "/" in model:
+        cfg["model_hf_repo"] = model
+        cfg["model_name"] = model.split("/")[-1]
+    else:
+        cfg["model_name"] = model
     cfg["quantization"] = args.quantization
     cfg["host"] = args.host
+    if args.model_id:
+        cfg["model_hf_repo"] = args.model_id
     if args.shard_start is not None:
         cfg["shard_start"] = args.shard_start
     if args.shard_end is not None:
@@ -255,7 +242,7 @@ def _cmd_start(args) -> int:
             tracker_url=cfg["tracker_url"],
             port=cfg["port"],
             model=cfg["model_name"],
-            model_id=cfg.get("model_hf_repo") or None,
+            model_id=cfg.get("model_hf_repo"),
             shard_start=cfg.get("shard_start"),
             shard_end=cfg.get("shard_end"),
             quantization=cfg["quantization"].replace("bf16", "bfloat16"),
@@ -301,8 +288,7 @@ def main() -> None:
     )
 
     # Flags that apply to the no-subcommand (default) path
-    parser.add_argument("--model", metavar="MODEL", help="Model name or HuggingFace repo ID to serve")
-    parser.add_argument("--model-id", metavar="MODEL", help="Alias for --model (catalog name or HuggingFace repo)")
+    parser.add_argument("--model", metavar="HF_REPO", help="HuggingFace repo ID to serve")
     parser.add_argument("--quantization", "-q", choices=["bf16", "int8", "nf4", "bfloat16"],
                         help="Quantization level")
     parser.add_argument("--download-dir", metavar="PATH", help="Model download directory")
@@ -343,8 +329,8 @@ def main() -> None:
     start_cmd = subparsers.add_parser("start", help="Start node (legacy flags)")
     start_cmd.add_argument("--tracker")
     start_cmd.add_argument("--port", type=int)
-    start_cmd.add_argument("--model", help="Model name or HuggingFace repo ID")
-    start_cmd.add_argument("--model-id", help="Alias for --model (catalog name or HuggingFace repo)")
+    start_cmd.add_argument("--model")
+    start_cmd.add_argument("--model-id", help="HuggingFace repo ID")
     start_cmd.add_argument("--shard-start", type=int)
     start_cmd.add_argument("--shard-end", type=int)
     start_cmd.add_argument("--quantization", choices=["auto", "bfloat16", "int8", "nf4", "bf16"], default="auto")

packages/node/meshnet_node/startup.py

@@ -205,6 +205,21 @@ def _max_assignable_layers(
     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(
     memory_mb: int,
     memory_source: str,
@@ -331,9 +346,6 @@ def _attach_relay_bridge(node: StubNodeServer | TorchNodeServer, bridge: RelayHt
     node.stop = _stop_with_bridge  # type: ignore[method-assign]
 
 
-_PENDING_NODE_ID = "pending"
-
-
 def _start_heartbeat(
     tracker_url: str,
     node_id: str,
@@ -371,8 +383,6 @@ def _start_heartbeat(
         try:
             resp = _post_json(f"{tracker_url}/v1/nodes/register", register_payload)
             node_id = resp.get("node_id", node_id)
-            if node_ref is not None:
-                setattr(node_ref, "tracker_node_id", node_id)
             return True
         except Exception:
             return False
@@ -424,7 +434,7 @@ def _start_heartbeat(
     def _loop() -> None:
         nonlocal node_id
         hb_url = f"{tracker_url}/v1/nodes/{node_id}/heartbeat"
-        outage_streak = 1 if node_id == _PENDING_NODE_ID else 0
+        outage_streak = 0  # consecutive intervals where tracker was unreachable
 
         while True:
             time.sleep(interval)
@@ -479,34 +489,6 @@ def _start_heartbeat(
     return t
 
 
-def _register_with_tracker(
-    tracker_url: str,
-    reg_payload: dict,
-    node: Any,
-    start_time: float,
-) -> str | None:
-    """Register with the tracker, or start background retries when it is unreachable."""
-    try:
-        reg_resp = _post_json(f"{tracker_url}/v1/nodes/register", reg_payload)
-        tracker_node_id = str(reg_resp.get("node_id") or "?")
-        setattr(node, "tracker_node_id", tracker_node_id)
-        print(f"  Registered with tracker — node ID: {tracker_node_id}", flush=True)
-        _start_heartbeat(tracker_url, tracker_node_id, reg_payload, node_ref=node, start_time=start_time)
-        return tracker_node_id
-    except Exception as exc:
-        setattr(node, "tracker_node_id", None)
-        print(f"  Warning: tracker registration failed: {exc}", flush=True)
-        print("  [node] will retry registration in the background", flush=True)
-        _start_heartbeat(
-            tracker_url,
-            _PENDING_NODE_ID,
-            reg_payload,
-            node_ref=node,
-            start_time=start_time,
-        )
-        return None
-
-
 def _warn_virtual_network_ip(ip: str | None) -> None:
     """Print a warning when the auto-detected advertise IP is in a known virtual-network range.
 
@@ -666,11 +648,8 @@ def run_startup(
     if probationary_line is not None:
         print(f"  {probationary_line}", flush=True)
 
-    pinned_shard_start = shard_start
-    pinned_shard_end = shard_end
-    user_pinned_shard = pinned_shard_start is not None or pinned_shard_end is not None
-
     if model_id:  # treat "" the same as None — no explicit model given
+        user_pinned_shard = shard_start is not None or shard_end is not None
         full_sources: list[dict] = []
         # Auto-detect shard range from model config if not explicitly provided
         if shard_start is None or shard_end is None:
@@ -734,11 +713,7 @@ def run_startup(
         _node_start_time = time.monotonic()
         actual_port = node.start()
         total_layers = getattr(getattr(node, "backend", None), "total_layers", None)
-        if isinstance(total_layers, int) and total_layers > 0:
-            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}"
+        shard_label = _format_shard_label(shard_start, shard_end, total_layers)
         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}"
@@ -782,9 +757,16 @@ def run_startup(
             **registration_capabilities,
             **relay_fields,
         }
-        tracker_node_id = _register_with_tracker(
-            tracker_url, reg_payload, node, _node_start_time,
-        )
+        tracker_node_id: str | None = None
+        try:
+            reg_resp = _post_json(f"{tracker_url}/v1/nodes/register", reg_payload)
+            tracker_node_id = str(reg_resp.get("node_id") or "?")
+            setattr(node, "tracker_node_id", tracker_node_id)
+            print(f"  Registered with tracker — node ID: {tracker_node_id}", flush=True)
+            _start_heartbeat(tracker_url, tracker_node_id, reg_payload, node_ref=node, start_time=_node_start_time)
+        except Exception as exc:
+            setattr(node, "tracker_node_id", None)
+            print(f"  Warning: tracker registration failed: {exc}", flush=True)
 
         print(
             f"\n{'=' * 32}\n"
@@ -802,8 +784,8 @@ def run_startup(
             flush=True,
         )
         return node
-    if user_pinned_shard and not model:
-        raise ValueError("--shard-start / --shard-end require --model")
+    if shard_start is not None or shard_end is not None:
+        raise ValueError("--shard-start / --shard-end require --model-id")
 
     # 3a. Auto-join: query tracker for network-wide HF model assignment.
     # Skipped when the user explicitly requested a model — the shard-assignment
@@ -910,17 +892,27 @@ def run_startup(
             **registration_capabilities,
             **relay_fields,
         }
-        tracker_node_id = _register_with_tracker(
-            tracker_url, auto_reg_payload, node, _node_start_time,
+        tracker_node_id = None
+        try:
+            reg_resp = _post_json(f"{tracker_url}/v1/nodes/register", auto_reg_payload)
+            tracker_node_id = str(reg_resp.get("node_id") or "?")
+            setattr(node, "tracker_node_id", tracker_node_id)
+            print(f"  Registered with tracker — node ID: {tracker_node_id}", flush=True)
+            _start_heartbeat(tracker_url, tracker_node_id, auto_reg_payload, node_ref=node, start_time=_node_start_time)
+        except Exception as exc:
+            setattr(node, "tracker_node_id", None)
+            print(f"  Warning: tracker registration failed: {exc}", flush=True)
+        shard_label = _format_shard_label(
+            assigned_shard_start,
+            assigned_shard_end,
+            assigned_num_layers,
         )
-        shard_count = assigned_shard_end - assigned_shard_start + 1
         print(
             f"\n{'=' * 32}\n"
             f"meshnet-node ready (auto-joined)\n"
             f"  Wallet:       {address}\n"
             f"  Model ID:     {assigned_hf_repo}\n"
-            f"  Shard:        layers {assigned_shard_start}–{assigned_shard_end} "
-            f"({shard_count} of {assigned_num_layers})\n"
+            f"  Shard:        {shard_label}\n"
             f"  {_shard_budget_line(memory_budget_mb, memory_budget_source, assigned_num_layers, quantization)}\n"
             f"  Quantization: {quantization}\n"
             f"  Endpoint:     {endpoint}\n"
@@ -955,25 +947,23 @@ def run_startup(
         print(f"  ERROR: Cannot reach tracker at {tracker_url}: {exc}", file=sys.stderr, flush=True)
         raise
 
-    shard_start = assignment["shard_start"]
-    shard_end = assignment["shard_end"]
-    if user_pinned_shard:
-        if pinned_shard_start is not None:
-            shard_start = pinned_shard_start
-        if pinned_shard_end is not None:
-            shard_end = pinned_shard_end
+    shard_start: int = assignment["shard_start"]
+    shard_end: int = assignment["shard_end"]
     assigned_model: str = assignment.get("model", model)
     hf_repo: str | None = assignment.get("hf_repo")
     peers: list[dict] = assignment.get("peers", [])
     model_sources: list[dict] = [] if tracker_source_disabled else assignment.get("model_sources", [])
     assignment_bytes_per_layer = _assignment_bytes_per_layer(assignment, quantization)
-    if user_pinned_shard:
-        print(
-            f"  Shard: layers {shard_start}-{shard_end} of {assigned_model} (pinned)",
-            flush=True,
-        )
-    else:
-        print(f"  Shard: layers {shard_start}-{shard_end} of {assigned_model}", flush=True)
+    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
+    )
+    print(
+        f"  Shard: {_format_shard_label(shard_start, shard_end, assigned_total_layers, model_name=assigned_model)}",
+        flush=True,
+    )
 
     # 4. Download shard
     print("Downloading shard...", flush=True)
@@ -1035,7 +1025,6 @@ def run_startup(
         "hardware_profile": hw,
         "wallet_address": address,
         "score": 1.0,
-        "managed_assignment": not user_pinned_shard,
         **registration_capabilities,
         **relay_fields,
     }
@@ -1055,12 +1044,18 @@ def run_startup(
     hw_str = device.upper()
     if gpu_name:
         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(
         f"\n{'=' * 32}\n"
         f"meshnet-node ready\n"
         f"  Wallet:   {address}\n"
-        f"  Shard:    layers {shard_start}-{shard_end} ({assigned_model})\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:    {shard_label}\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"  Node ID:  {node_id}\n"
         f"  Hardware: {hw_str}\n"

packages/node/meshnet_node/torch_server.py

@@ -368,7 +368,11 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler):
         if stream:
             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:
                 payload = backend.encode_prompt(current_text)
             except Exception as exc:
@@ -386,6 +390,21 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler):
             if stream_emit is not None:
                 stream_emit(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)
         if stream_emit is not None:

packages/tracker/meshnet_tracker/dashboard.html

@@ -88,6 +88,9 @@
   .status-processing { color:var(--accent); }
   .status-failed { color:var(--bad); }
   .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>
 </head>
 <body>
@@ -350,6 +353,13 @@ function buildCallWallStates(events) {
       rec.elapsed = f.elapsed_seconds;
       rec.stream = f.stream;
       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") {
       rec.status = "failed";
       rec.model = rec.model || f.model || f.route_model || "?";
@@ -379,7 +389,7 @@ function renderCallWall(consoleData, stats) {
   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);
+    else if (rec.status === "complete" || rec.status === "failed" || rec.status === "canceled") 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));
@@ -401,10 +411,13 @@ function renderCallWall(consoleData, stats) {
     `</div>`;
 
   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 note = rec.warn || (rec.route ? short(String(rec.route), 28) : "");
-      return [
+      const row = [
         `<span class="${statusCls}">${esc(rec.status)}</span>`,
         `<span class="num">${esc(callWallAgeSeconds(rec, nowSec).toFixed(1))}s</span>`,
         esc(short(rec.model || "?", 28)),
@@ -414,6 +427,12 @@ function renderCallWall(consoleData, stats) {
         `<span class="num">${esc(String(callWallMaxQueue(rec)))}</span>`,
         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 {
     html += '<div class="empty">no in-flight requests</div>';
@@ -422,10 +441,16 @@ function renderCallWall(consoleData, stats) {
   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 statusCls = rec.status === "failed"
+      ? "status-failed"
+      : rec.status === "canceled"
+        ? "status-canceled"
+        : "status-complete";
     const detail = rec.status === "failed"
       ? esc(short(rec.error || "?", 40))
-      : (f.stream ? "stream" : "json");
+      : rec.status === "canceled"
+        ? "canceled"
+        : (f.stream ? "stream" : "json");
     return [
       new Date((e.ts || 0) * 1000).toLocaleTimeString(),
       `<span class="${statusCls}">${esc(rec.status)}</span>`,
@@ -437,7 +462,7 @@ function renderCallWall(consoleData, stats) {
       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
     ? table(["time", "status", "model", "request", "tps", "tokens", "sec", "detail"], historyRows)
     : '<div class="empty">no completed requests yet</div>';
@@ -956,6 +981,23 @@ async function renderAdminPanel() {
   $("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() {
   $("self-url").textContent = location.host;
   const [raft, map, stats, models, consoleData, adminData] = await Promise.all([

packages/tracker/meshnet_tracker/server.py

@@ -35,6 +35,7 @@ import urllib.parse
 import urllib.request
 import uuid
 from collections import deque
+from dataclasses import dataclass, field
 from importlib.resources import files
 from pathlib import Path
 from typing import Any
@@ -51,6 +52,7 @@ from .raft import RaftNode
 
 
 _CONSOLE_LIMIT = 300
+_PROXY_PROGRESS_LOG_INTERVAL = 5.0
 
 
 def _preset_price_keys(name: str, preset: dict) -> set[str]:
@@ -1413,6 +1415,10 @@ def _relay_http_request_frames(
     headers: dict[str, str],
     timeout: float = 310.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
     response frames until a terminal one (US-036).
@@ -1430,6 +1436,14 @@ def _relay_http_request_frames(
     deadline = time.monotonic() + timeout
     try:
         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({
                 "request_id": request_id,
                 "method": "POST",
@@ -1438,6 +1452,8 @@ def _relay_http_request_frames(
                 "body": body.decode(errors="replace"),
             }))
             while True:
+                if cancel_event is not None and cancel_event.is_set():
+                    return
                 remaining = deadline - time.monotonic()
                 if remaining <= 0:
                     return
@@ -2076,7 +2092,15 @@ def _registration_ban_error(contracts: Any | None, wallet_address: str | 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 = {
         "ts": time.time(),
         "level": level,
@@ -2088,10 +2112,88 @@ def _tracker_log(server: "_TrackerHTTPServer", level: str, message: str, **field
         },
     }
     with server.console_lock:
-        server.console_events.append(event)
-    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)
+        if update_console_key is not None:
+            updated = False
+            for existing in reversed(server.console_events):
+                if (
+                    existing.get("message") == message
+                    and existing.get("fields", {}).get("request_id") == update_console_key
+                ):
+                    existing["ts"] = event["ts"]
+                    existing["fields"] = event["fields"]
+                    updated = True
+                    break
+            if not updated:
+                server.console_events.append(event)
+        else:
+            server.console_events.append(event)
+    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(
@@ -2108,10 +2210,21 @@ def _tracker_log_proxy_progress(
 ) -> None:
     elapsed = time.monotonic() - started
     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(
         server,
         "info",
         "proxy progress",
+        stdout=stdout,
+        update_console_key=request_id,
         request_id=request_id,
         model=model,
         route_model=route_model,
@@ -2209,6 +2322,8 @@ class _TrackerHTTPServer(socketserver.ThreadingMixIn, http.server.HTTPServer):
         self.models_dir = models_dir
         self.console_events = deque(maxlen=_CONSOLE_LIMIT)
         self.console_lock = threading.Lock()
+        self.active_proxies: dict[str, _ActiveProxyContext] = {}
+        self.active_proxies_lock = threading.Lock()
 
 
 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":
             self._handle_heartbeat(parts[3])
             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.end_headers()
 
@@ -2886,6 +3012,9 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
             if inflight_recorded:
                 _record_proxy_inflight(server, inflight_nodes, -1)
                 inflight_recorded = False
+            _unregister_active_proxy(server, request_id)
+
+        proxy_ctx = _register_active_proxy(server, request_id)
 
         _tracker_log(
             server,
@@ -2930,13 +3059,34 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
                 direct_endpoint=target_url,
             )
             started = time.monotonic()
+            relay_ws_holder: list[Any] = []
             frames = _relay_http_request_frames(
                 node.relay_addr,
                 path="/v1/chat/completions",
                 body=raw_body,
                 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)
+            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"):
                 # Streamed response (US-036): forward SSE chunks as they arrive
                 # and run the same token accounting as the direct stream path.
@@ -2945,6 +3095,8 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
                     model, route_model, route_nodes, api_key, node_work,
                     request_body=body,
                     request_id=request_id,
+                    proxy_ctx=proxy_ctx,
+                    finish_proxy_inflight=finish_proxy_inflight,
                 )
                 finish_proxy_inflight()
                 return
@@ -2959,6 +3111,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
                         in_tokens, out_tokens = 0, 0
                     tokens = in_tokens + out_tokens
                     self._record_observed_throughput(model, route_model, tokens, elapsed, route_nodes)
+                    _clear_proxy_progress_log_state(server, request_id)
                     _tracker_log(
                         server,
                         "info",
@@ -2989,7 +3142,69 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
 
         try:
             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)
         except urllib.error.HTTPError as exc:
             # Relay error status + body from node
@@ -3002,9 +3217,11 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
                 self.wfile.write(err_body)
             except BrokenPipeError:
                 pass
+            _clear_proxy_progress_log_state(server, request_id)
             finish_proxy_inflight()
             return
         except Exception as exc:
+            _clear_proxy_progress_log_state(server, request_id)
             if node.relay_addr:
                 _tracker_log(
                     server,
@@ -3045,8 +3262,15 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
                 client_gone = False
                 try:
                     while True:
-                        line = upstream.readline()
+                        if proxy_ctx.cancel_event.is_set():
+                            break
+                        try:
+                            line = upstream.readline()
+                        except TimeoutError:
+                            continue
                         if not line:
+                            if proxy_ctx.cancel_event.is_set():
+                                break
                             break
                         if not client_gone:
                             try:
@@ -3071,6 +3295,22 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
                             stream_usage = usage
                 except (BrokenPipeError, ConnectionResetError):
                     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
                 # Bill even on client disconnect — the nodes did the work.
                 # Observed stream chunks are authoritative for the upper bound;
@@ -3082,6 +3322,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
                     model, route_model, in_tokens + out_tokens, elapsed, route_nodes
                 )
                 tokens = in_tokens + out_tokens
+                _clear_proxy_progress_log_state(server, request_id)
                 _tracker_log(
                     server,
                     "info",
@@ -3113,6 +3354,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
                     in_tokens, out_tokens = 0, 0
                 tokens = in_tokens + out_tokens
                 self._record_observed_throughput(model, route_model, tokens, elapsed, route_nodes)
+                _clear_proxy_progress_log_state(server, request_id)
                 _tracker_log(
                     server,
                     "info",
@@ -3272,6 +3514,9 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
         node_work: list,
         request_body: dict,
         request_id: str,
+        *,
+        proxy_ctx: _ActiveProxyContext | None = None,
+        finish_proxy_inflight: Any = None,
     ) -> None:
         """Forward a streamed relay response (US-036) to the client as SSE,
         billing with the same accounting as the direct stream path."""
@@ -3285,6 +3530,8 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
         observed_stream_tokens = 0
         client_gone = False
         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 ""
             if not chunk:
                 continue
@@ -3312,6 +3559,26 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
                     )
                 if usage is not None:
                     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
         in_tokens, out_tokens = _stream_billable_split(
             observed_stream_tokens, stream_usage, request_body
@@ -3320,6 +3587,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
             model, route_model, in_tokens + out_tokens, elapsed, route_nodes
         )
         tokens = in_tokens + out_tokens
+        _clear_proxy_progress_log_state(server, request_id)
         _tracker_log(
             server,
             "info",
@@ -3765,6 +4033,68 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
             events = [dict(event) for event in server.console_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):
         server: _TrackerHTTPServer = self.server  # type: ignore[assignment]
         if not self._require_role("admin"):

tests/test_mining_cli.py

@@ -388,107 +388,6 @@ def test_legacy_start_treats_repo_model_as_model_id(monkeypatch):
     assert captured["model_id"] == "Qwen/Qwen2.5-0.5B-Instruct"
 
 
-def test_legacy_start_catalog_model_with_pinned_shards(monkeypatch):
-    """Catalog model names accept --shard-start/--shard-end without --model-id."""
-    from meshnet_node.cli import main
-
-    captured = {}
-
-    def fake_run_startup(*args, **kwargs):
-        captured.update(kwargs)
-        class _FakeNode:
-            chat_completion_count = 0
-            def stop(self): pass
-        return _FakeNode()
-
-    monkeypatch.setattr(sys, "argv", [
-        "meshnet-node", "start",
-        "--tracker", "http://192.168.0.179:8080",
-        "--model", "Qwen3.6-35B-A3B",
-        "--shard-start", "0",
-        "--shard-end", "44",
-        "--port", "0",
-    ])
-
-    with patch("meshnet_node.startup.run_startup", side_effect=fake_run_startup):
-        with patch("time.sleep", side_effect=KeyboardInterrupt):
-            try:
-                main()
-            except SystemExit as exc:
-                assert exc.code == 0
-
-    assert captured["model"] == "Qwen3.6-35B-A3B"
-    assert captured["model_id"] is None
-    assert captured["shard_start"] == 0
-    assert captured["shard_end"] == 44
-
-
-def test_legacy_start_model_id_alias_for_catalog_name(monkeypatch):
-    """--model-id with a catalog name routes through the tracker preset path."""
-    from meshnet_node.cli import main
-
-    captured = {}
-
-    def fake_run_startup(*args, **kwargs):
-        captured.update(kwargs)
-        class _FakeNode:
-            chat_completion_count = 0
-            def stop(self): pass
-        return _FakeNode()
-
-    monkeypatch.setattr(sys, "argv", [
-        "meshnet-node", "start",
-        "--tracker", "http://192.168.0.179:8080",
-        "--model-id", "Qwen3.6-35B-A3B",
-        "--port", "0",
-    ])
-
-    with patch("meshnet_node.startup.run_startup", side_effect=fake_run_startup):
-        with patch("time.sleep", side_effect=KeyboardInterrupt):
-            try:
-                main()
-            except SystemExit as exc:
-                assert exc.code == 0
-
-    assert captured["model"] == "Qwen3.6-35B-A3B"
-    assert captured["model_id"] is None
-
-
-def test_legacy_start_hf_repo_with_pinned_shards(monkeypatch):
-    """HF repo --model with pinned shards still enters the torch startup path."""
-    from meshnet_node.cli import main
-
-    captured = {}
-
-    def fake_run_startup(*args, **kwargs):
-        captured.update(kwargs)
-        class _FakeNode:
-            chat_completion_count = 0
-            def stop(self): pass
-        return _FakeNode()
-
-    monkeypatch.setattr(sys, "argv", [
-        "meshnet-node", "start",
-        "--tracker", "http://192.168.0.179:8081",
-        "--model", "Qwen/Qwen2.5-0.5B-Instruct",
-        "--shard-start", "12",
-        "--shard-end", "23",
-        "--port", "0",
-    ])
-
-    with patch("meshnet_node.startup.run_startup", side_effect=fake_run_startup):
-        with patch("time.sleep", side_effect=KeyboardInterrupt):
-            try:
-                main()
-            except SystemExit as exc:
-                assert exc.code == 0
-
-    assert captured["model"] == "Qwen2.5-0.5B-Instruct"
-    assert captured["model_id"] == "Qwen/Qwen2.5-0.5B-Instruct"
-    assert captured["shard_start"] == 12
-    assert captured["shard_end"] == 23
-
-
 def test_legacy_start_falls_back_to_env_tracker_and_model(monkeypatch):
     """`meshnet-node start` uses env defaults when tracker/model flags are omitted."""
     import importlib

tests/test_node_startup.py

@@ -5,10 +5,8 @@ import io
 import os
 import sys
 import tarfile
-import threading
 import time
 import types
-import urllib.error
 import urllib.request
 from pathlib import Path
 
@@ -1118,7 +1116,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["max_loaded_shards"] == 2
     output = capsys.readouterr().out
-    assert "Shard:        layers 0–23; 24 of 24" in output
+    assert "Shard:        layers 0–23 (24 of 24)" in output
     assert "Node ID:      node-test-123" in output
 
 
@@ -1605,120 +1603,6 @@ def test_preset_model_startup_starts_heartbeat(tmp_path, monkeypatch):
         tracker.stop()
 
 
-def test_preset_model_startup_honors_pinned_shard_range(tmp_path, monkeypatch):
-    """Explicit --shard-start/--shard-end override tracker auto-assignment."""
-    import meshnet_node.startup as startup_mod
-
-    monkeypatch.setattr(
-        startup_mod,
-        "detect_hardware",
-        lambda: {"device": "cpu", "gpu_name": None, "vram_mb": 0, "ram_mb": 16 * 1024},
-    )
-    heartbeat_calls = []
-    monkeypatch.setattr(
-        startup_mod,
-        "_start_heartbeat",
-        lambda *args, **kwargs: heartbeat_calls.append((args, kwargs)),
-    )
-
-    tracker = TrackerServer(model_presets={"stub-model": {"layers_start": 0, "layers_end": 15}})
-    tracker_port = tracker.start()
-    tracker_url = f"http://127.0.0.1:{tracker_port}"
-    try:
-        node = run_startup(
-            tracker_url=tracker_url,
-            model="stub-model",
-            shard_start=0,
-            shard_end=5,
-            wallet_path=tmp_path / "wallet.json",
-            cache_dir=tmp_path / "shards",
-        )
-        try:
-            assert len(heartbeat_calls) == 1
-            args, kwargs = heartbeat_calls[0]
-            reg_payload = args[2]
-            assert reg_payload["shard_start"] == 0
-            assert reg_payload["shard_end"] == 5
-            assert reg_payload["managed_assignment"] is False
-        finally:
-            node.stop()
-    finally:
-        tracker.stop()
-
-
-def test_torch_startup_retries_registration_when_tracker_unreachable(
-    tmp_path,
-    monkeypatch,
-):
-    """Failed initial registration should start background retry, not stay unregistered."""
-    import meshnet_node.startup as startup_mod
-
-    class FakeBackend:
-        total_layers = 24
-
-    class FakeTorchNodeServer:
-        def __init__(self, **kwargs):
-            self.backend = FakeBackend()
-            self.port = None
-            self.chat_completion_count = 0
-            self.tracker_node_id = None
-
-        def start(self):
-            self.port = 7000
-            return self.port
-
-        def stop(self):
-            pass
-
-    monkeypatch.setattr(
-        startup_mod,
-        "detect_hardware",
-        lambda: {"device": "cuda", "gpu_name": "Test GPU", "vram_mb": 8192, "ram_mb": 16 * 1024},
-    )
-    monkeypatch.setattr(startup_mod, "TorchNodeServer", FakeTorchNodeServer)
-    monkeypatch.setattr(
-        startup_mod,
-        "_detect_num_layers",
-        lambda *_args, **_kwargs: 24,
-    )
-
-    heartbeat_calls = []
-    monkeypatch.setattr(
-        startup_mod,
-        "_start_heartbeat",
-        lambda *args, **kwargs: heartbeat_calls.append((args, kwargs)) or threading.Thread(),
-    )
-
-    register_calls = {"count": 0}
-
-    def flaky_register(url, payload):
-        register_calls["count"] += 1
-        raise urllib.error.URLError("connection refused")
-
-    monkeypatch.setattr(startup_mod, "_post_json", flaky_register)
-
-    tracker = TrackerServer()
-    tracker_port = tracker.start()
-    tracker_url = f"http://127.0.0.1:{tracker_port}"
-    try:
-        node = run_startup(
-            tracker_url=tracker_url,
-            model_id="Qwen/Qwen2.5-0.5B-Instruct",
-            wallet_path=tmp_path / "wallet.json",
-        )
-        try:
-            assert register_calls["count"] == 1
-            assert node.tracker_node_id is None
-            assert len(heartbeat_calls) == 1
-            args, kwargs = heartbeat_calls[0]
-            assert args[1] == startup_mod._PENDING_NODE_ID
-            assert kwargs["node_ref"] is node
-        finally:
-            node.stop()
-    finally:
-        tracker.stop()
-
-
 def test_real_model_startup_registers_downloaded_inventory_without_checksum(
     tmp_path,
     monkeypatch,

tests/test_tracker_routing.py

@@ -5,6 +5,7 @@ import json
 import threading
 import time
 import urllib.error
+import urllib.parse
 import urllib.request
 
 import pytest
@@ -501,6 +502,100 @@ def test_tracker_logs_stream_progress_before_request_completes():
         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():
     """The documented qwen2.5-0.5b alias resolves a full HF repo registration."""
     tracker = TrackerServer()