Compare commits
5 Commits
worktree-a
...
worktree-a
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
dbf856f497 | ||
|
|
753f553766 | ||
|
|
85c13e4e82 | ||
|
|
0152d5ed99 | ||
|
|
8ea70ff6a0 |
@@ -138,3 +138,41 @@ The OpenRouter model is set via `--model` (e.g. `openai/gpt-4o`, `mistralai/mist
|
||||
- `ralph_progress.py run-next --agent custom --agent-cmd ./my-agent.sh` runs a task via custom script
|
||||
- `python -m pytest` passes from repo root
|
||||
- Commit only this story's changes
|
||||
|
||||
---
|
||||
|
||||
## Part 4: Per-story metadata in dashboard
|
||||
|
||||
Every story in the dashboard gets an optional second line showing what is known about it:
|
||||
|
||||
```
|
||||
✓ US-012 12 — Real PyTorch model backend
|
||||
codex · done · "Added model_backend.py with bitsandbytes NF4 support"
|
||||
⚡ US-014 14 — Tracker-as-first-layer-node
|
||||
codex · in-progress · worktree: ../AI-worktree-us-014
|
||||
→ US-015 15 — Ralph agent-agnostic runner
|
||||
claude · in-progress · worktree: ../AI-worktree-us-015
|
||||
```
|
||||
|
||||
### Sources
|
||||
|
||||
| Data | Source |
|
||||
|------|--------|
|
||||
| Active worktree path | `git worktree list --porcelain` — match branches `feat/<story-id>` |
|
||||
| Agent (active story) | `.ralph-tui/agent-config.json` → `agent` field |
|
||||
| Agent (completed story) | `.ralph-tui/session.json` → `agentPlugin`; or `completionNotes` text |
|
||||
| Last output summary | `story.completionNotes` (done); latest `git log -1 --format=%s feat/<id>` (worktree); last lines of most recent `.ralph-tui/iterations/<hash>_<date>_<id>.log` (in-progress) |
|
||||
|
||||
### Behaviour
|
||||
|
||||
- Metadata line is omitted if there is nothing to show (no agent, no worktree, no notes)
|
||||
- Default: shown (verbose). `--compact` flag suppresses it for a tighter one-line-per-story view
|
||||
- `_active_worktrees() -> dict[story_id, rel_path]` helper, called once per dashboard render
|
||||
- `_story_meta(story, worktrees, session) -> str | None` helper returns the joined metadata string
|
||||
|
||||
### Additional acceptance criteria
|
||||
|
||||
- `ralph_progress.py show` displays worktree path for any story whose branch exists as `feat/<story-id>`
|
||||
- Agent name appears next to in-progress and recently-completed stories
|
||||
- `completionNotes` from prd.json appears as the summary for done stories
|
||||
- `--compact` suppresses metadata lines
|
||||
|
||||
@@ -346,13 +346,14 @@
|
||||
"Commit only this story's changes"
|
||||
],
|
||||
"priority": 14,
|
||||
"passes": false,
|
||||
"passes": true,
|
||||
"notes": "Source issue: .scratch/distributed-inference-network/issues/14-tracker-as-node.md",
|
||||
"dependsOn": [
|
||||
"US-012",
|
||||
"US-013"
|
||||
],
|
||||
"status": "open"
|
||||
"status": "done",
|
||||
"completionNotes": "Implemented: (1) Tracker GET /v1/tracker-nodes/<model> endpoint returning nodes registered with tracker_mode=true at shard_start==0. (2) StubNodeServer and TorchNodeServer now accept tracker_mode/tracker_url params; when tracker_mode=True, /v1/chat/completions is served alongside /forward. (3) TorchNodeServer auto-detects tracker mode when shard_start==0. (4) Gateway _handle_chat_completions checks for tracker-nodes first via _get_tracker_nodes(), proxies round-robin if found, falls back to existing direct pipeline if none (backward compat). (5) CLI --tracker-mode and --tracker-url flags added. (6) Integration test: 2 tracker-nodes + 2 mid-shard nodes for gpt2; 10 requests; round-robin verified (5/5 split); all responses valid OpenAI format. All 78 tests pass."
|
||||
},
|
||||
{
|
||||
"id": "US-015",
|
||||
|
||||
@@ -56,6 +56,7 @@ class _GatewayHTTPServer(http.server.HTTPServer):
|
||||
self.minimum_stake = minimum_stake
|
||||
self.cost_per_layer_token_lamport = cost_per_layer_token_lamport
|
||||
self.last_binary_chunk_responses: list[_BinaryActivation] = []
|
||||
self.request_count: int = 0
|
||||
|
||||
|
||||
class _GatewayHandler(http.server.BaseHTTPRequestHandler):
|
||||
@@ -243,11 +244,28 @@ class _GatewayHandler(http.server.BaseHTTPRequestHandler):
|
||||
|
||||
def _handle_chat_completions(self):
|
||||
server: _GatewayHTTPServer = self.server # type: ignore[assignment]
|
||||
body = self._read_json_body()
|
||||
if body is None:
|
||||
# Read raw bytes first so we can proxy them if tracker-nodes are available
|
||||
length = int(self.headers.get("Content-Length", 0))
|
||||
raw_body = self.rfile.read(length)
|
||||
try:
|
||||
body = json.loads(raw_body or b"{}")
|
||||
except (json.JSONDecodeError, ValueError):
|
||||
self._send_json_error(400, "invalid JSON body")
|
||||
return
|
||||
if not isinstance(body, dict):
|
||||
self._send_json_error(400, "JSON body must be an object")
|
||||
return
|
||||
streaming = bool(body.get("stream", False))
|
||||
|
||||
model = str(body.get("model", "stub-model"))
|
||||
tracker_nodes = _get_tracker_nodes(server, model)
|
||||
if tracker_nodes:
|
||||
# Proxy to a tracker-node (round-robin by request count)
|
||||
target = tracker_nodes[server.request_count % len(tracker_nodes)]
|
||||
server.request_count += 1
|
||||
return self._proxy_to_tracker_node(target, raw_body)
|
||||
|
||||
# Fallback: use existing direct pipeline (backward compat)
|
||||
streaming = bool(body.get("stream", False))
|
||||
try:
|
||||
completion = self._build_completion(body)
|
||||
except _ModelUnavailable as exc:
|
||||
@@ -266,6 +284,37 @@ class _GatewayHandler(http.server.BaseHTTPRequestHandler):
|
||||
else:
|
||||
self._send_json(200, completion)
|
||||
|
||||
def _proxy_to_tracker_node(self, url: str, body_bytes: bytes) -> None:
|
||||
"""Forward a raw request body to a tracker-node and stream the response back."""
|
||||
target_url = f"{url}/v1/chat/completions"
|
||||
req = urllib.request.Request(
|
||||
target_url,
|
||||
data=body_bytes,
|
||||
headers={"Content-Type": "application/json"},
|
||||
method="POST",
|
||||
)
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=30.0) as r:
|
||||
content_type = r.headers.get("Content-Type", "application/json")
|
||||
resp_body = r.read()
|
||||
status = r.status
|
||||
except urllib.error.HTTPError as exc:
|
||||
resp_body = exc.read()
|
||||
self.send_response(exc.code)
|
||||
self.send_header("Content-Type", "application/json")
|
||||
self.send_header("Content-Length", str(len(resp_body)))
|
||||
self.end_headers()
|
||||
self.wfile.write(resp_body)
|
||||
return
|
||||
except Exception as exc:
|
||||
self._send_json_error(503, f"tracker-node unreachable: {exc}")
|
||||
return
|
||||
self.send_response(status)
|
||||
self.send_header("Content-Type", content_type)
|
||||
self.send_header("Content-Length", str(len(resp_body)))
|
||||
self.end_headers()
|
||||
self.wfile.write(resp_body)
|
||||
|
||||
def _handle_meshnet_request(self) -> None:
|
||||
body = self._read_json_body()
|
||||
if body is None:
|
||||
@@ -767,6 +816,17 @@ def _get_json(url: str, timeout: float = 5.0) -> dict:
|
||||
return json.loads(r.read())
|
||||
|
||||
|
||||
def _get_tracker_nodes(server: _GatewayHTTPServer, model: str) -> list[str]:
|
||||
"""Return tracker-node endpoint URLs for this model, or empty list on any failure."""
|
||||
if server.tracker_url is None:
|
||||
return []
|
||||
try:
|
||||
resp = _get_json(f"{server.tracker_url}/v1/tracker-nodes/{urllib.parse.quote(model)}")
|
||||
return [n["endpoint"] for n in resp.get("tracker_nodes", [])]
|
||||
except Exception:
|
||||
return []
|
||||
|
||||
|
||||
class GatewayServer:
|
||||
"""HTTP gateway that routes /v1/chat/completions through an ordered inference route.
|
||||
|
||||
|
||||
@@ -39,6 +39,16 @@ def main() -> None:
|
||||
"--advertise-host",
|
||||
help="Reachable host/IP to advertise to the tracker (defaults to FQDN when binding 0.0.0.0)",
|
||||
)
|
||||
start_cmd.add_argument(
|
||||
"--tracker-mode",
|
||||
action="store_true",
|
||||
help="Enable client-facing /v1/chat/completions (auto-enabled when shard-start=0)",
|
||||
)
|
||||
start_cmd.add_argument(
|
||||
"--tracker-url",
|
||||
default=None,
|
||||
help="Tracker URL for route selection (used in tracker mode)",
|
||||
)
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
|
||||
@@ -2,6 +2,7 @@
|
||||
|
||||
import http.server
|
||||
import json
|
||||
import time
|
||||
import threading
|
||||
import urllib.parse
|
||||
from pathlib import Path
|
||||
@@ -93,6 +94,8 @@ class _StubHTTPServer(http.server.HTTPServer):
|
||||
response_prefix: str,
|
||||
model: str,
|
||||
shard_path: Path | None,
|
||||
tracker_mode: bool = False,
|
||||
tracker_url: str | None = None,
|
||||
):
|
||||
super().__init__(addr, handler)
|
||||
self.shard_start = shard_start
|
||||
@@ -103,6 +106,9 @@ class _StubHTTPServer(http.server.HTTPServer):
|
||||
self.shard_path = shard_path
|
||||
self.received_activations: bool = False
|
||||
self.forward_chunk_count: int = 0
|
||||
self.tracker_mode: bool = tracker_mode
|
||||
self.tracker_url: str | None = tracker_url
|
||||
self.chat_completion_count: int = 0
|
||||
|
||||
|
||||
class _StubHandler(http.server.BaseHTTPRequestHandler):
|
||||
@@ -110,10 +116,13 @@ class _StubHandler(http.server.BaseHTTPRequestHandler):
|
||||
pass
|
||||
|
||||
def do_POST(self):
|
||||
server: _StubHTTPServer = self.server # type: ignore[assignment]
|
||||
if self.path == "/v1/infer":
|
||||
self._handle_infer()
|
||||
elif self.path == "/forward":
|
||||
self._handle_forward()
|
||||
elif self.path == "/v1/chat/completions" and server.tracker_mode:
|
||||
self._handle_chat_completions()
|
||||
else:
|
||||
self.send_response(404)
|
||||
self.end_headers()
|
||||
@@ -126,6 +135,82 @@ class _StubHandler(http.server.BaseHTTPRequestHandler):
|
||||
self.send_response(404)
|
||||
self.end_headers()
|
||||
|
||||
def _send_json(self, status: int, data: dict) -> None:
|
||||
payload = json.dumps(data).encode()
|
||||
self.send_response(status)
|
||||
self.send_header("Content-Type", "application/json")
|
||||
self.send_header("Content-Length", str(len(payload)))
|
||||
self.end_headers()
|
||||
self.wfile.write(payload)
|
||||
|
||||
def _handle_chat_completions(self) -> None:
|
||||
server: _StubHTTPServer = self.server # type: ignore[assignment]
|
||||
length = int(self.headers.get("Content-Length", 0))
|
||||
try:
|
||||
body = json.loads(self.rfile.read(length) or b"{}")
|
||||
except (json.JSONDecodeError, ValueError):
|
||||
self._send_json(400, {"error": "invalid JSON body"})
|
||||
return
|
||||
if not isinstance(body, dict):
|
||||
self._send_json(400, {"error": "JSON body must be an object"})
|
||||
return
|
||||
server.chat_completion_count += 1
|
||||
streaming = bool(body.get("stream", False))
|
||||
model = str(body.get("model", server.model))
|
||||
messages = body.get("messages", [])
|
||||
last_content = ""
|
||||
if isinstance(messages, list) and messages:
|
||||
last = messages[-1]
|
||||
if isinstance(last, dict):
|
||||
last_content = str(last.get("content", ""))
|
||||
text = f"{server.response_prefix} {last_content}"
|
||||
if streaming:
|
||||
self._send_sse_response(text, model)
|
||||
else:
|
||||
created = int(time.time())
|
||||
self._send_json(200, {
|
||||
"id": "chatcmpl-stub",
|
||||
"object": "chat.completion",
|
||||
"created": created,
|
||||
"model": model,
|
||||
"choices": [{
|
||||
"index": 0,
|
||||
"message": {"role": "assistant", "content": text},
|
||||
"finish_reason": "stop",
|
||||
}],
|
||||
"usage": {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0},
|
||||
})
|
||||
|
||||
def _send_sse_response(self, text: str, model: str) -> None:
|
||||
chunk_id = "chatcmpl-stub"
|
||||
created = int(time.time())
|
||||
self.send_response(200)
|
||||
self.send_header("Content-Type", "text/event-stream; charset=utf-8")
|
||||
self.send_header("Cache-Control", "no-cache")
|
||||
self.end_headers()
|
||||
|
||||
def _emit(data: str) -> None:
|
||||
self.wfile.write(f"data: {data}\n\n".encode())
|
||||
self.wfile.flush()
|
||||
|
||||
_emit(json.dumps({
|
||||
"id": chunk_id, "object": "chat.completion.chunk", "created": created,
|
||||
"model": model,
|
||||
"choices": [{"index": 0, "delta": {"role": "assistant", "content": ""}, "finish_reason": None}],
|
||||
}))
|
||||
_emit(json.dumps({
|
||||
"id": chunk_id, "object": "chat.completion.chunk", "created": created,
|
||||
"model": model,
|
||||
"choices": [{"index": 0, "delta": {"content": text}, "finish_reason": None}],
|
||||
}))
|
||||
_emit(json.dumps({
|
||||
"id": chunk_id, "object": "chat.completion.chunk", "created": created,
|
||||
"model": model,
|
||||
"choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}],
|
||||
}))
|
||||
self.wfile.write(b"data: [DONE]\n\n")
|
||||
self.wfile.flush()
|
||||
|
||||
def _handle_shard_download(self, parsed: urllib.parse.ParseResult):
|
||||
server: _StubHTTPServer = self.server # type: ignore[assignment]
|
||||
params = urllib.parse.parse_qs(parsed.query)
|
||||
@@ -246,6 +331,7 @@ class StubNodeServer:
|
||||
shard_start / shard_end define which transformer layer range this node owns.
|
||||
is_last_shard controls whether the node returns a text response (True) or
|
||||
activation tensors (False) after processing its shard.
|
||||
tracker_mode enables the /v1/chat/completions endpoint for head-shard nodes.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
@@ -258,6 +344,8 @@ class StubNodeServer:
|
||||
response_prefix: str = "stub response to:",
|
||||
model: str = "stub-model",
|
||||
shard_path: Path | None = None,
|
||||
tracker_mode: bool = False,
|
||||
tracker_url: str | None = None,
|
||||
):
|
||||
self._host = host
|
||||
self._requested_port = port
|
||||
@@ -269,6 +357,8 @@ class StubNodeServer:
|
||||
self._response_prefix = response_prefix
|
||||
self._model = model
|
||||
self._shard_path = shard_path
|
||||
self._tracker_mode = tracker_mode
|
||||
self._tracker_url = tracker_url
|
||||
self._server: _StubHTTPServer | None = None
|
||||
self._thread: threading.Thread | None = None
|
||||
self.port: int | None = None
|
||||
@@ -283,6 +373,11 @@ class StubNodeServer:
|
||||
"""Number of binary /forward chunks handled since this node was started."""
|
||||
return self._server.forward_chunk_count if self._server is not None else 0
|
||||
|
||||
@property
|
||||
def chat_completion_count(self) -> int:
|
||||
"""Number of /v1/chat/completions requests handled since this node was started."""
|
||||
return self._server.chat_completion_count if self._server is not None else 0
|
||||
|
||||
def start(self) -> int:
|
||||
if self._server is not None:
|
||||
raise RuntimeError("StubNodeServer is already running")
|
||||
@@ -296,6 +391,8 @@ class StubNodeServer:
|
||||
self._response_prefix,
|
||||
self._model,
|
||||
self._shard_path,
|
||||
self._tracker_mode,
|
||||
self._tracker_url,
|
||||
)
|
||||
self.port = self._server.server_address[1]
|
||||
self._thread = threading.Thread(target=self._server.serve_forever, daemon=True)
|
||||
|
||||
@@ -6,6 +6,11 @@ import http.server
|
||||
import json
|
||||
import sys
|
||||
import threading
|
||||
import time
|
||||
import urllib.error
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
import uuid
|
||||
|
||||
from .model_backend import (
|
||||
InsufficientVRAMError,
|
||||
@@ -24,11 +29,20 @@ from .server import (
|
||||
|
||||
|
||||
class _TorchHTTPServer(http.server.HTTPServer):
|
||||
def __init__(self, addr, handler, backend: TorchModelShard):
|
||||
def __init__(
|
||||
self,
|
||||
addr,
|
||||
handler,
|
||||
backend: TorchModelShard,
|
||||
tracker_mode: bool = False,
|
||||
tracker_url: str | None = None,
|
||||
):
|
||||
super().__init__(addr, handler)
|
||||
self.backend = backend
|
||||
self.received_activations = False
|
||||
self.forward_chunk_count = 0
|
||||
self.tracker_mode = tracker_mode
|
||||
self.tracker_url = tracker_url
|
||||
|
||||
|
||||
class _TorchHandler(http.server.BaseHTTPRequestHandler):
|
||||
@@ -36,10 +50,13 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler):
|
||||
pass
|
||||
|
||||
def do_POST(self):
|
||||
server: _TorchHTTPServer = self.server # type: ignore[assignment]
|
||||
if self.path == "/forward":
|
||||
self._handle_forward()
|
||||
elif self.path == "/v1/infer":
|
||||
self._handle_infer()
|
||||
elif self.path == "/v1/chat/completions" and server.tracker_mode:
|
||||
self._handle_chat_completions()
|
||||
else:
|
||||
self.send_response(404)
|
||||
self.end_headers()
|
||||
@@ -190,6 +207,152 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler):
|
||||
self.end_headers()
|
||||
self.wfile.write(payload)
|
||||
|
||||
def _handle_chat_completions(self) -> None:
|
||||
server: _TorchHTTPServer = self.server # type: ignore[assignment]
|
||||
body = self._read_json_body()
|
||||
if body is None:
|
||||
return
|
||||
messages = body.get("messages", [])
|
||||
stream = bool(body.get("stream", False))
|
||||
model = str(body.get("model", ""))
|
||||
prompt = " ".join(
|
||||
str(m.get("content", ""))
|
||||
for m in messages
|
||||
if isinstance(m, dict) and m.get("role") == "user"
|
||||
)
|
||||
try:
|
||||
payload = server.backend.encode_prompt(prompt)
|
||||
except Exception as exc:
|
||||
self._send_json(500, {"error": f"encode_prompt failed: {exc}"})
|
||||
return
|
||||
remaining_route = self._get_remaining_route(model)
|
||||
result_text = self._run_downstream_pipeline(payload, remaining_route)
|
||||
self._send_openai_response(result_text, model, stream)
|
||||
|
||||
def _get_remaining_route(self, model: str) -> list[str]:
|
||||
server: _TorchHTTPServer = self.server # type: ignore[assignment]
|
||||
if server.tracker_url is None:
|
||||
return []
|
||||
try:
|
||||
url = f"{server.tracker_url}/v1/route?model={urllib.parse.quote(model)}"
|
||||
with urllib.request.urlopen(url, timeout=5.0) as r:
|
||||
route_resp = json.loads(r.read())
|
||||
route = route_resp.get("route", [])
|
||||
# Skip the first node in the route (self) since we're already the head
|
||||
return list(route[1:])
|
||||
except Exception:
|
||||
return []
|
||||
|
||||
def _run_downstream_pipeline(self, payload: object, route: list[str]) -> str:
|
||||
server: _TorchHTTPServer = self.server # type: ignore[assignment]
|
||||
if not route:
|
||||
# Single-node mode: decode tail locally if we're the tail
|
||||
if server.backend.is_tail:
|
||||
try:
|
||||
tensor = server.backend.torch.frombuffer(
|
||||
bytearray(payload.body), # type: ignore[union-attr]
|
||||
dtype=server.backend.torch.bfloat16,
|
||||
).reshape(payload.shape).to(server.backend.device) # type: ignore[union-attr]
|
||||
return server.backend.decode_tail(tensor)
|
||||
except Exception as exc:
|
||||
return f"decode error: {exc}"
|
||||
return ""
|
||||
|
||||
session = str(uuid.uuid4())
|
||||
shape = payload.shape # type: ignore[union-attr]
|
||||
attn_mask = payload.attention_mask_header # type: ignore[union-attr]
|
||||
pos_ids = payload.position_ids_header # type: ignore[union-attr]
|
||||
current_body = payload.body # type: ignore[union-attr]
|
||||
current_shape = shape
|
||||
current_attn = attn_mask
|
||||
current_pos = pos_ids
|
||||
|
||||
for hop_index, node_url in enumerate(route):
|
||||
headers: dict[str, str] = {
|
||||
"Content-Type": "application/octet-stream",
|
||||
"X-Meshnet-Wire": _WIRE_VERSION,
|
||||
"X-Meshnet-Shape": ",".join(str(d) for d in current_shape),
|
||||
"X-Meshnet-Dtype": "bfloat16",
|
||||
"X-Meshnet-Session": session,
|
||||
"X-Meshnet-Chunk-Index": "0",
|
||||
"X-Meshnet-Chunk-Total": "1",
|
||||
"X-Meshnet-Hop-Index": str(hop_index),
|
||||
}
|
||||
if current_attn:
|
||||
headers["X-Meshnet-Attn-Mask"] = current_attn
|
||||
if current_pos:
|
||||
headers["X-Meshnet-Position-Ids"] = current_pos
|
||||
req = urllib.request.Request(
|
||||
f"{node_url}/forward",
|
||||
data=current_body,
|
||||
headers=headers,
|
||||
method="POST",
|
||||
)
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=10.0) as r:
|
||||
resp_body = r.read()
|
||||
resp_headers = {k.lower(): v for k, v in r.headers.items()}
|
||||
except Exception as exc:
|
||||
return f"pipeline error at {node_url}: {exc}"
|
||||
content_type = resp_headers.get("content-type", "")
|
||||
if "application/json" in content_type:
|
||||
try:
|
||||
data = json.loads(resp_body)
|
||||
return str(data.get("text", ""))
|
||||
except json.JSONDecodeError:
|
||||
return resp_body.decode("utf-8", errors="replace")
|
||||
# Binary activation — update and forward to next node
|
||||
shape_header = resp_headers.get("x-meshnet-shape", ",".join(str(d) for d in current_shape))
|
||||
current_shape = _parse_shape(shape_header)
|
||||
current_body = resp_body
|
||||
current_attn = resp_headers.get("x-meshnet-attn-mask")
|
||||
current_pos = resp_headers.get("x-meshnet-position-ids")
|
||||
return ""
|
||||
|
||||
def _send_openai_response(self, text: str, model: str, stream: bool) -> None:
|
||||
chunk_id = "chatcmpl-node"
|
||||
created = int(time.time())
|
||||
if not stream:
|
||||
self._send_json(200, {
|
||||
"id": chunk_id,
|
||||
"object": "chat.completion",
|
||||
"created": created,
|
||||
"model": model,
|
||||
"choices": [{
|
||||
"index": 0,
|
||||
"message": {"role": "assistant", "content": text},
|
||||
"finish_reason": "stop",
|
||||
}],
|
||||
"usage": {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0},
|
||||
})
|
||||
return
|
||||
self.send_response(200)
|
||||
self.send_header("Content-Type", "text/event-stream; charset=utf-8")
|
||||
self.send_header("Cache-Control", "no-cache")
|
||||
self.end_headers()
|
||||
|
||||
def _emit(data: str) -> None:
|
||||
self.wfile.write(f"data: {data}\n\n".encode())
|
||||
self.wfile.flush()
|
||||
|
||||
_emit(json.dumps({
|
||||
"id": chunk_id, "object": "chat.completion.chunk", "created": created,
|
||||
"model": model,
|
||||
"choices": [{"index": 0, "delta": {"role": "assistant", "content": ""}, "finish_reason": None}],
|
||||
}))
|
||||
_emit(json.dumps({
|
||||
"id": chunk_id, "object": "chat.completion.chunk", "created": created,
|
||||
"model": model,
|
||||
"choices": [{"index": 0, "delta": {"content": text}, "finish_reason": None}],
|
||||
}))
|
||||
_emit(json.dumps({
|
||||
"id": chunk_id, "object": "chat.completion.chunk", "created": created,
|
||||
"model": model,
|
||||
"choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}],
|
||||
}))
|
||||
self.wfile.write(b"data: [DONE]\n\n")
|
||||
self.wfile.flush()
|
||||
|
||||
|
||||
class TorchNodeServer:
|
||||
"""HTTP server backed by a HuggingFace causal language model shard."""
|
||||
@@ -203,6 +366,8 @@ class TorchNodeServer:
|
||||
shard_end: int = 6,
|
||||
quantization: str = "bfloat16",
|
||||
backend: TorchModelShard | None = None,
|
||||
tracker_mode: bool | None = None,
|
||||
tracker_url: str | None = None,
|
||||
) -> None:
|
||||
self._host = host
|
||||
self._requested_port = port
|
||||
@@ -212,6 +377,9 @@ class TorchNodeServer:
|
||||
shard_end,
|
||||
quantization,
|
||||
)
|
||||
# Auto-detect tracker mode: enabled when shard_start == 0 or explicitly set
|
||||
self._tracker_mode = tracker_mode if tracker_mode is not None else (shard_start == 0)
|
||||
self._tracker_url = tracker_url
|
||||
self._server: _TorchHTTPServer | None = None
|
||||
self._thread: threading.Thread | None = None
|
||||
self.port: int | None = None
|
||||
@@ -235,6 +403,8 @@ class TorchNodeServer:
|
||||
(self._host, self._requested_port),
|
||||
_TorchHandler,
|
||||
self._backend,
|
||||
self._tracker_mode,
|
||||
self._tracker_url,
|
||||
)
|
||||
self.port = self._server.server_address[1]
|
||||
self._thread = threading.Thread(target=self._server.serve_forever, daemon=True)
|
||||
|
||||
@@ -31,28 +31,51 @@ from typing import Any
|
||||
|
||||
|
||||
DEFAULT_MODEL_PRESETS: dict[str, dict] = {
|
||||
"stub-model": {"layers_start": 0, "layers_end": 31},
|
||||
"stub-model": {
|
||||
"layers_start": 0,
|
||||
"layers_end": 31,
|
||||
"bytes_per_layer": {"bfloat16": 30 * 1024 * 1024},
|
||||
},
|
||||
"openai-community/gpt2": {
|
||||
"layers_start": 0,
|
||||
"layers_end": 11,
|
||||
"bytes_per_layer": {"bfloat16": 30 * 1024 * 1024, "int8": 15 * 1024 * 1024, "nf4": 8 * 1024 * 1024},
|
||||
},
|
||||
}
|
||||
|
||||
DEFAULT_VRAM_BYTES = 8 * 1024 * 1024 * 1024
|
||||
DEFAULT_RAM_BYTES = 16 * 1024 * 1024 * 1024
|
||||
DEFAULT_QUANTIZATIONS = ["bfloat16"]
|
||||
DEFAULT_BENCHMARK_TOKENS_PER_SEC = 1.0
|
||||
|
||||
|
||||
class _NodeEntry:
|
||||
__slots__ = (
|
||||
"node_id", "endpoint", "shard_start", "shard_end",
|
||||
"model", "shard_checksum", "hardware_profile", "wallet_address",
|
||||
"score", "last_heartbeat",
|
||||
"score", "vram_bytes", "ram_bytes", "quantizations",
|
||||
"benchmark_tokens_per_sec", "quantization", "managed_assignment",
|
||||
"pending_directives", "last_heartbeat", "tracker_mode",
|
||||
)
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
node_id: str,
|
||||
endpoint: str,
|
||||
shard_start: int,
|
||||
shard_end: int,
|
||||
shard_start: int | None,
|
||||
shard_end: int | None,
|
||||
model: str | None,
|
||||
shard_checksum: str | None,
|
||||
hardware_profile: dict,
|
||||
wallet_address: str | None,
|
||||
score: float,
|
||||
vram_bytes: int = DEFAULT_VRAM_BYTES,
|
||||
ram_bytes: int = DEFAULT_RAM_BYTES,
|
||||
quantizations: list[str] | None = None,
|
||||
benchmark_tokens_per_sec: float = DEFAULT_BENCHMARK_TOKENS_PER_SEC,
|
||||
quantization: str | None = None,
|
||||
managed_assignment: bool = False,
|
||||
tracker_mode: bool = False,
|
||||
) -> None:
|
||||
self.node_id = node_id
|
||||
self.endpoint = endpoint
|
||||
@@ -63,6 +86,14 @@ class _NodeEntry:
|
||||
self.hardware_profile = hardware_profile
|
||||
self.wallet_address = wallet_address
|
||||
self.score = score
|
||||
self.vram_bytes = vram_bytes
|
||||
self.ram_bytes = ram_bytes
|
||||
self.quantizations = quantizations or list(DEFAULT_QUANTIZATIONS)
|
||||
self.benchmark_tokens_per_sec = benchmark_tokens_per_sec
|
||||
self.quantization = quantization
|
||||
self.managed_assignment = managed_assignment
|
||||
self.tracker_mode = tracker_mode
|
||||
self.pending_directives: list[dict] = []
|
||||
self.last_heartbeat: float = time.monotonic()
|
||||
|
||||
|
||||
@@ -72,7 +103,10 @@ def _select_route(
|
||||
required_end: int,
|
||||
) -> tuple[list[_NodeEntry], str]:
|
||||
"""Greedy interval-cover. Returns (ordered route, error_message)."""
|
||||
candidates = sorted(nodes, key=lambda n: (n.shard_start, -n.shard_end))
|
||||
candidates = sorted(
|
||||
[node for node in nodes if node.shard_start is not None and node.shard_end is not None],
|
||||
key=lambda n: (n.shard_start, -n.shard_end), # type: ignore[operator]
|
||||
)
|
||||
route: list[_NodeEntry] = []
|
||||
covered_up_to = required_start - 1
|
||||
|
||||
@@ -105,6 +139,8 @@ def _coverage_percentage(
|
||||
(
|
||||
(max(required_start, node.shard_start), min(required_end, node.shard_end))
|
||||
for node in nodes
|
||||
if node.shard_start is not None
|
||||
and node.shard_end is not None
|
||||
if node.shard_end >= required_start and node.shard_start <= required_end
|
||||
),
|
||||
key=lambda interval: interval[0],
|
||||
@@ -122,6 +158,197 @@ def _coverage_percentage(
|
||||
return round((covered / required_layers) * 100, 2)
|
||||
|
||||
|
||||
def _preset_layer_bounds(preset: dict) -> tuple[int, int]:
|
||||
start = int(preset.get("layers_start", 0))
|
||||
if "layers_end" in preset:
|
||||
return start, int(preset["layers_end"])
|
||||
return start, start + int(preset["total_layers"]) - 1
|
||||
|
||||
|
||||
def _preset_bytes_per_layer(preset: dict) -> dict[str, int]:
|
||||
raw = preset.get("bytes_per_layer", preset.get("bytes_per_layer_at_quant", {}))
|
||||
if isinstance(raw, dict) and raw:
|
||||
return {str(quant): int(value) for quant, value in raw.items()}
|
||||
return {"bfloat16": 30 * 1024 * 1024}
|
||||
|
||||
|
||||
def _node_quantization(node: _NodeEntry, preset: dict) -> str:
|
||||
bytes_per_layer = _preset_bytes_per_layer(preset)
|
||||
if node.quantization in bytes_per_layer:
|
||||
return node.quantization
|
||||
for quantization in node.quantizations:
|
||||
if quantization in bytes_per_layer:
|
||||
return quantization
|
||||
return next(iter(bytes_per_layer))
|
||||
|
||||
|
||||
def _node_layer_capacity(node: _NodeEntry, preset: dict) -> int:
|
||||
bytes_per_layer = _preset_bytes_per_layer(preset)
|
||||
quantization = _node_quantization(node, preset)
|
||||
layer_bytes = bytes_per_layer[quantization]
|
||||
if layer_bytes <= 0:
|
||||
return 0
|
||||
return int((node.vram_bytes * 0.8) // layer_bytes)
|
||||
|
||||
|
||||
def _coverage_map(
|
||||
nodes: list[_NodeEntry],
|
||||
required_start: int,
|
||||
required_end: int,
|
||||
) -> list[dict]:
|
||||
layer_counts = []
|
||||
for layer in range(required_start, required_end + 1):
|
||||
count = 0
|
||||
for node in nodes:
|
||||
if node.shard_start is None or node.shard_end is None:
|
||||
continue
|
||||
if node.shard_start <= layer <= node.shard_end:
|
||||
count += 1
|
||||
layer_counts.append((layer, count))
|
||||
|
||||
coverage: list[dict] = []
|
||||
for layer, count in layer_counts:
|
||||
if coverage and coverage[-1]["node_count"] == count and coverage[-1]["end_layer"] == layer - 1:
|
||||
coverage[-1]["end_layer"] = layer
|
||||
else:
|
||||
coverage.append({"start_layer": layer, "end_layer": layer, "node_count": count})
|
||||
return coverage
|
||||
|
||||
|
||||
def _coverage_gaps(coverage: list[dict]) -> list[tuple[int, int]]:
|
||||
return [
|
||||
(segment["start_layer"], segment["end_layer"])
|
||||
for segment in coverage
|
||||
if segment["node_count"] == 0
|
||||
]
|
||||
|
||||
|
||||
def _load_directive(node: _NodeEntry, model: str, start: int, end: int, quantization: str) -> dict:
|
||||
return {
|
||||
"action": "LOAD_SHARD",
|
||||
"model": model,
|
||||
"start_layer": start,
|
||||
"end_layer": end,
|
||||
"shard_start": start,
|
||||
"shard_end": end,
|
||||
"quantization": quantization,
|
||||
}
|
||||
|
||||
|
||||
def _drop_directive(node: _NodeEntry, model: str, start: int, end: int, quantization: str) -> dict:
|
||||
return {
|
||||
"action": "DROP_SHARD",
|
||||
"model": model,
|
||||
"start_layer": start,
|
||||
"end_layer": end,
|
||||
"shard_start": start,
|
||||
"shard_end": end,
|
||||
"quantization": quantization,
|
||||
}
|
||||
|
||||
|
||||
def _purge_expired_nodes_locked(server: "_TrackerHTTPServer") -> list[str]:
|
||||
now = time.monotonic()
|
||||
expired_ids = [
|
||||
node_id for node_id, entry in server.registry.items()
|
||||
if (now - entry.last_heartbeat) > server.heartbeat_timeout
|
||||
]
|
||||
for node_id in expired_ids:
|
||||
del server.registry[node_id]
|
||||
if expired_ids:
|
||||
_rebalance_all_locked(server)
|
||||
return expired_ids
|
||||
|
||||
|
||||
def _rebalance_model_locked(server: "_TrackerHTTPServer", model: str) -> None:
|
||||
preset = server.model_presets.get(model)
|
||||
if preset is None:
|
||||
return
|
||||
required_start, required_end = _preset_layer_bounds(preset)
|
||||
total_layers = required_end - required_start + 1
|
||||
model_nodes = [node for node in server.registry.values() if node.model == model]
|
||||
managed_nodes = [node for node in model_nodes if node.managed_assignment]
|
||||
if not managed_nodes:
|
||||
return
|
||||
|
||||
previous_ranges = {
|
||||
node.node_id: (node.shard_start, node.shard_end, node.quantization)
|
||||
for node in managed_nodes
|
||||
}
|
||||
for node in managed_nodes:
|
||||
node.shard_start = None
|
||||
node.shard_end = None
|
||||
|
||||
managed_nodes.sort(
|
||||
key=lambda node: (
|
||||
-node.benchmark_tokens_per_sec,
|
||||
-_node_layer_capacity(node, preset),
|
||||
node.node_id,
|
||||
)
|
||||
)
|
||||
base_nodes = [node for node in model_nodes if not node.managed_assignment]
|
||||
coverage = _coverage_map(base_nodes, required_start, required_end)
|
||||
gaps = _coverage_gaps(coverage)
|
||||
if not gaps:
|
||||
gaps = [(required_start, required_end)]
|
||||
|
||||
eligible_nodes = [
|
||||
node for node in managed_nodes
|
||||
if _node_layer_capacity(node, preset) > 0
|
||||
]
|
||||
node_index = 0
|
||||
for gap_start, gap_end in gaps:
|
||||
cursor = gap_start
|
||||
while cursor <= gap_end and node_index < len(eligible_nodes):
|
||||
node = eligible_nodes[node_index]
|
||||
remaining_layers = gap_end - cursor + 1
|
||||
remaining_nodes_after = len(eligible_nodes) - node_index - 1
|
||||
capacity = min(
|
||||
_node_layer_capacity(node, preset),
|
||||
total_layers,
|
||||
max(1, remaining_layers - remaining_nodes_after),
|
||||
)
|
||||
if capacity <= 0:
|
||||
node_index += 1
|
||||
continue
|
||||
quantization = _node_quantization(node, preset)
|
||||
node.quantization = quantization
|
||||
node.shard_start = cursor
|
||||
node.shard_end = min(gap_end, cursor + capacity - 1)
|
||||
cursor = node.shard_end + 1
|
||||
node_index += 1
|
||||
|
||||
for node in managed_nodes:
|
||||
previous_start, previous_end, previous_quantization = previous_ranges[node.node_id]
|
||||
current_range = (node.shard_start, node.shard_end, node.quantization)
|
||||
if node.shard_start is None or node.shard_end is None or current_range == previous_ranges[node.node_id]:
|
||||
continue
|
||||
if previous_start is not None and previous_end is not None:
|
||||
node.pending_directives.append(
|
||||
_drop_directive(
|
||||
node,
|
||||
model,
|
||||
previous_start,
|
||||
previous_end,
|
||||
previous_quantization or _node_quantization(node, preset),
|
||||
)
|
||||
)
|
||||
node.pending_directives.append(
|
||||
_load_directive(
|
||||
node,
|
||||
model,
|
||||
node.shard_start,
|
||||
node.shard_end,
|
||||
node.quantization or _node_quantization(node, preset),
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def _rebalance_all_locked(server: "_TrackerHTTPServer") -> None:
|
||||
for model in list(server.model_presets):
|
||||
_rebalance_model_locked(server, model)
|
||||
|
||||
|
||||
def _registration_ban_error(contracts: Any | None, wallet_address: str | None) -> str | None:
|
||||
if contracts is None or not wallet_address:
|
||||
return None
|
||||
@@ -177,13 +404,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
|
||||
|
||||
def _purge_expired_nodes(self) -> None:
|
||||
server: _TrackerHTTPServer = self.server # type: ignore[assignment]
|
||||
now = time.monotonic()
|
||||
expired_ids = [
|
||||
node_id for node_id, entry in server.registry.items()
|
||||
if (now - entry.last_heartbeat) > server.heartbeat_timeout
|
||||
]
|
||||
for node_id in expired_ids:
|
||||
del server.registry[node_id]
|
||||
_purge_expired_nodes_locked(server)
|
||||
|
||||
def do_POST(self):
|
||||
if self.path == "/v1/nodes/register":
|
||||
@@ -207,6 +428,12 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
|
||||
self._handle_assign(parsed)
|
||||
elif parsed.path == "/v1/models":
|
||||
self._handle_models()
|
||||
elif parsed.path.startswith("/v1/coverage/"):
|
||||
model = urllib.parse.unquote(parsed.path.removeprefix("/v1/coverage/"))
|
||||
self._handle_coverage(model)
|
||||
elif parsed.path.startswith("/v1/tracker-nodes/"):
|
||||
model = urllib.parse.unquote(parsed.path.removeprefix("/v1/tracker-nodes/"))
|
||||
self._handle_tracker_nodes(model)
|
||||
else:
|
||||
self.send_response(404)
|
||||
self.end_headers()
|
||||
@@ -225,10 +452,13 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
|
||||
data = []
|
||||
for name, preset in server.model_presets.items():
|
||||
model_nodes = [node for node in alive if node.model == name]
|
||||
if not model_nodes:
|
||||
continue
|
||||
required_start, required_end = _preset_layer_bounds(preset)
|
||||
coverage = _coverage_percentage(
|
||||
model_nodes,
|
||||
preset["layers_start"],
|
||||
preset["layers_end"],
|
||||
required_start,
|
||||
required_end,
|
||||
)
|
||||
data.append({
|
||||
"id": name,
|
||||
@@ -239,6 +469,58 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
|
||||
})
|
||||
self._send_json(200, {"object": "list", "data": data})
|
||||
|
||||
def _handle_coverage(self, model: str):
|
||||
server: _TrackerHTTPServer = self.server # type: ignore[assignment]
|
||||
preset = server.model_presets.get(model)
|
||||
if preset is None:
|
||||
self._send_json(404, {"error": f"unknown model preset: {model!r}"})
|
||||
return
|
||||
required_start, required_end = _preset_layer_bounds(preset)
|
||||
with server.lock:
|
||||
self._purge_expired_nodes()
|
||||
alive = [node for node in server.registry.values() if node.model == model]
|
||||
if server.contracts is not None:
|
||||
alive = [
|
||||
node for node in alive
|
||||
if not node.wallet_address or not server.contracts.registry.get_wallet(node.wallet_address).banned
|
||||
]
|
||||
coverage = _coverage_map(alive, required_start, required_end)
|
||||
self._send_json(200, {"model": model, "coverage": coverage})
|
||||
|
||||
def _handle_tracker_nodes(self, model: str):
|
||||
"""Return nodes registered with tracker_mode=True whose shard starts at layer 0."""
|
||||
server: _TrackerHTTPServer = self.server # type: ignore[assignment]
|
||||
preset = server.model_presets.get(model)
|
||||
if preset is None:
|
||||
self._send_json(404, {"error": f"unknown model preset: {model!r}"})
|
||||
return
|
||||
required_start, _ = _preset_layer_bounds(preset)
|
||||
with server.lock:
|
||||
self._purge_expired_nodes()
|
||||
alive = [node for node in server.registry.values() if node.model == model]
|
||||
if server.contracts is not None:
|
||||
alive = [
|
||||
node for node in alive
|
||||
if not node.wallet_address or not server.contracts.registry.get_wallet(node.wallet_address).banned
|
||||
]
|
||||
tracker_nodes = [
|
||||
node for node in alive
|
||||
if node.shard_start is not None
|
||||
and node.shard_start == required_start
|
||||
and node.tracker_mode
|
||||
]
|
||||
self._send_json(200, {
|
||||
"model": model,
|
||||
"tracker_nodes": [
|
||||
{
|
||||
"node_id": node.node_id,
|
||||
"endpoint": node.endpoint,
|
||||
"benchmark_tokens_per_sec": node.benchmark_tokens_per_sec,
|
||||
}
|
||||
for node in tracker_nodes
|
||||
],
|
||||
})
|
||||
|
||||
def _handle_register(self):
|
||||
server: _TrackerHTTPServer = self.server # type: ignore[assignment]
|
||||
body = self._read_json_body()
|
||||
@@ -254,15 +536,26 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
|
||||
self._send_json(400, {"error": "endpoint must be an http(s) URL"})
|
||||
return
|
||||
|
||||
shard_start: int | None
|
||||
shard_end: int | None
|
||||
explicit_shard = "shard_start" in body or "shard_end" in body
|
||||
if explicit_shard:
|
||||
try:
|
||||
shard_start = int(body["shard_start"])
|
||||
shard_end = int(body["shard_end"])
|
||||
except (KeyError, TypeError, ValueError):
|
||||
self._send_json(400, {"error": "shard_start and shard_end must be numeric"})
|
||||
return
|
||||
if shard_start < 0 or shard_end < 0 or shard_start > shard_end:
|
||||
self._send_json(400, {"error": "shard range must be non-negative and ordered"})
|
||||
return
|
||||
else:
|
||||
shard_start = None
|
||||
shard_end = None
|
||||
try:
|
||||
shard_start = int(body["shard_start"])
|
||||
shard_end = int(body["shard_end"])
|
||||
score = float(body.get("score", 1.0))
|
||||
except (KeyError, TypeError, ValueError):
|
||||
self._send_json(400, {"error": "shard_start, shard_end, and score must be numeric"})
|
||||
return
|
||||
if shard_start < 0 or shard_end < 0 or shard_start > shard_end:
|
||||
self._send_json(400, {"error": "shard range must be non-negative and ordered"})
|
||||
except (TypeError, ValueError):
|
||||
self._send_json(400, {"error": "score must be numeric"})
|
||||
return
|
||||
|
||||
hardware_profile = body.get("hardware_profile", {})
|
||||
@@ -279,6 +572,31 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
|
||||
if shard_checksum is not None and not isinstance(shard_checksum, str):
|
||||
self._send_json(400, {"error": "shard_checksum must be a string"})
|
||||
return
|
||||
try:
|
||||
vram_bytes = int(body.get("vram_bytes", DEFAULT_VRAM_BYTES))
|
||||
ram_bytes = int(body.get("ram_bytes", DEFAULT_RAM_BYTES))
|
||||
benchmark_tokens_per_sec = float(
|
||||
body.get("benchmark_tokens_per_sec", DEFAULT_BENCHMARK_TOKENS_PER_SEC)
|
||||
)
|
||||
except (TypeError, ValueError):
|
||||
self._send_json(400, {"error": "vram_bytes, ram_bytes, and benchmark_tokens_per_sec must be numeric"})
|
||||
return
|
||||
if vram_bytes < 0 or ram_bytes < 0 or benchmark_tokens_per_sec <= 0:
|
||||
self._send_json(400, {"error": "capability values must be positive"})
|
||||
return
|
||||
quantizations_body = body.get("quantizations", DEFAULT_QUANTIZATIONS)
|
||||
if not (
|
||||
isinstance(quantizations_body, list)
|
||||
and quantizations_body
|
||||
and all(isinstance(item, str) and item for item in quantizations_body)
|
||||
):
|
||||
self._send_json(400, {"error": "quantizations must be a non-empty string array"})
|
||||
return
|
||||
quantizations = list(quantizations_body)
|
||||
quantization = body.get("quantization")
|
||||
if quantization is not None and not isinstance(quantization, str):
|
||||
self._send_json(400, {"error": "quantization must be a string"})
|
||||
return
|
||||
wallet_address = body.get("wallet_address")
|
||||
if wallet_address is not None and not isinstance(wallet_address, str):
|
||||
self._send_json(400, {"error": "wallet_address must be a string"})
|
||||
@@ -288,6 +606,8 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
|
||||
self._send_json(403, {"error": ban_error})
|
||||
return
|
||||
|
||||
tracker_mode = bool(body.get("tracker_mode", False))
|
||||
|
||||
node_id = str(uuid.uuid4())
|
||||
entry = _NodeEntry(
|
||||
node_id=node_id,
|
||||
@@ -299,12 +619,27 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
|
||||
hardware_profile=hardware_profile,
|
||||
wallet_address=wallet_address,
|
||||
score=score,
|
||||
vram_bytes=vram_bytes,
|
||||
ram_bytes=ram_bytes,
|
||||
quantizations=quantizations,
|
||||
benchmark_tokens_per_sec=benchmark_tokens_per_sec,
|
||||
quantization=quantization,
|
||||
managed_assignment=not explicit_shard,
|
||||
tracker_mode=tracker_mode,
|
||||
)
|
||||
with server.lock:
|
||||
self._purge_expired_nodes()
|
||||
server.registry[node_id] = entry
|
||||
if entry.managed_assignment:
|
||||
_rebalance_model_locked(server, model)
|
||||
assignment_directive = entry.pending_directives[-1] if entry.pending_directives else None
|
||||
if assignment_directive is not None:
|
||||
entry.pending_directives.clear()
|
||||
|
||||
self._send_json(200, {"node_id": node_id})
|
||||
payload = {"node_id": node_id}
|
||||
if assignment_directive is not None:
|
||||
payload["directive"] = assignment_directive
|
||||
self._send_json(200, payload)
|
||||
|
||||
def _handle_heartbeat(self, node_id: str):
|
||||
server: _TrackerHTTPServer = self.server # type: ignore[assignment]
|
||||
@@ -315,7 +650,13 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
|
||||
self._send_json(404, {"error": "node not found"})
|
||||
return
|
||||
entry.last_heartbeat = time.monotonic()
|
||||
self._send_json(200, {})
|
||||
_rebalance_model_locked(server, entry.model or "stub-model")
|
||||
directives = list(entry.pending_directives)
|
||||
entry.pending_directives.clear()
|
||||
if directives:
|
||||
self._send_json(200, {"directives": directives})
|
||||
else:
|
||||
self._send_json(200, {})
|
||||
|
||||
def _handle_assign(self, parsed: urllib.parse.ParseResult):
|
||||
"""Return an optimal shard assignment for a node given its hardware profile.
|
||||
@@ -346,8 +687,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
|
||||
self._send_json(404, {"error": f"unknown model preset: {model!r}"})
|
||||
return
|
||||
|
||||
required_start: int = preset["layers_start"]
|
||||
required_end: int = preset["layers_end"]
|
||||
required_start, required_end = _preset_layer_bounds(preset)
|
||||
|
||||
with server.lock:
|
||||
self._purge_expired_nodes()
|
||||
@@ -369,7 +709,11 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
|
||||
|
||||
# Collect covered intervals sorted by start layer.
|
||||
covered = sorted(
|
||||
[(n.shard_start, n.shard_end) for n in alive],
|
||||
[
|
||||
(n.shard_start, n.shard_end)
|
||||
for n in alive
|
||||
if n.shard_start is not None and n.shard_end is not None
|
||||
],
|
||||
key=lambda t: t[0],
|
||||
)
|
||||
|
||||
@@ -421,8 +765,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
|
||||
self._send_json(404, {"error": f"unknown model preset: {model!r}"})
|
||||
return
|
||||
|
||||
required_start: int = preset["layers_start"]
|
||||
required_end: int = preset["layers_end"]
|
||||
required_start, required_end = _preset_layer_bounds(preset)
|
||||
|
||||
with server.lock:
|
||||
self._purge_expired_nodes()
|
||||
@@ -477,8 +820,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
|
||||
self._send_json(404, {"error": f"unknown model preset: {model!r}"})
|
||||
return
|
||||
|
||||
required_start: int = preset["layers_start"]
|
||||
required_end: int = preset["layers_end"]
|
||||
required_start, required_end = _preset_layer_bounds(preset)
|
||||
|
||||
with server.lock:
|
||||
self._purge_expired_nodes()
|
||||
@@ -530,6 +872,7 @@ class TrackerServer:
|
||||
host: str = "127.0.0.1",
|
||||
port: int = 0,
|
||||
heartbeat_timeout: float = 30.0,
|
||||
rebalance_interval: float = 30.0,
|
||||
model_presets: dict | None = None,
|
||||
contracts: Any | None = None,
|
||||
minimum_stake: int = 0,
|
||||
@@ -537,6 +880,7 @@ class TrackerServer:
|
||||
self._host = host
|
||||
self._requested_port = port
|
||||
self._heartbeat_timeout = heartbeat_timeout
|
||||
self._rebalance_interval = rebalance_interval
|
||||
self._model_presets: dict = (
|
||||
model_presets if model_presets is not None else dict(DEFAULT_MODEL_PRESETS)
|
||||
)
|
||||
@@ -546,6 +890,8 @@ class TrackerServer:
|
||||
self._lock = threading.Lock()
|
||||
self._server: _TrackerHTTPServer | None = None
|
||||
self._thread: threading.Thread | None = None
|
||||
self._rebalance_stop = threading.Event()
|
||||
self._rebalance_thread: threading.Thread | None = None
|
||||
self.port: int | None = None
|
||||
|
||||
def start(self) -> int:
|
||||
@@ -562,17 +908,33 @@ class TrackerServer:
|
||||
self._minimum_stake,
|
||||
)
|
||||
self.port = self._server.server_address[1]
|
||||
self._rebalance_stop.clear()
|
||||
self._thread = threading.Thread(target=self._server.serve_forever, daemon=True)
|
||||
self._thread.start()
|
||||
self._rebalance_thread = threading.Thread(target=self._rebalance_loop, daemon=True)
|
||||
self._rebalance_thread.start()
|
||||
return self.port
|
||||
|
||||
def _rebalance_loop(self) -> None:
|
||||
while not self._rebalance_stop.wait(self._rebalance_interval):
|
||||
server = self._server
|
||||
if server is None:
|
||||
return
|
||||
with self._lock:
|
||||
_purge_expired_nodes_locked(server)
|
||||
_rebalance_all_locked(server)
|
||||
|
||||
def stop(self) -> None:
|
||||
if self._server is None:
|
||||
return
|
||||
self._rebalance_stop.set()
|
||||
self._server.shutdown()
|
||||
self._server.server_close()
|
||||
if self._thread is not None:
|
||||
self._thread.join(timeout=1)
|
||||
if self._rebalance_thread is not None:
|
||||
self._rebalance_thread.join(timeout=1)
|
||||
self._server = None
|
||||
self._thread = None
|
||||
self._rebalance_thread = None
|
||||
self.port = None
|
||||
|
||||
@@ -92,6 +92,75 @@ def _default_agent() -> str:
|
||||
return cfg.get("agent") or DEFAULT_AGENT
|
||||
|
||||
|
||||
def _active_worktrees() -> dict[str, str]:
|
||||
"""Return {STORY-ID: relative-path} for worktrees on feat/<story-id> branches."""
|
||||
r = subprocess.run(
|
||||
["git", "worktree", "list", "--porcelain"],
|
||||
cwd=REPO_ROOT, text=True,
|
||||
stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False,
|
||||
)
|
||||
worktrees: dict[str, str] = {}
|
||||
current_path: str | None = None
|
||||
for line in r.stdout.splitlines():
|
||||
if line.startswith("worktree "):
|
||||
current_path = line[len("worktree "):].strip()
|
||||
elif line.startswith("branch refs/heads/feat/") and current_path:
|
||||
slug = line[len("branch refs/heads/feat/"):].strip() # e.g. "us-015"
|
||||
story_id = slug.upper() # "US-015"
|
||||
try:
|
||||
rel = str(Path(current_path).relative_to(REPO_ROOT))
|
||||
except ValueError:
|
||||
rel = current_path
|
||||
worktrees[story_id] = rel
|
||||
return worktrees
|
||||
|
||||
|
||||
def _story_last_commit(story_id: str) -> str:
|
||||
"""Latest commit subject on feat/<story-id> branch, or empty string."""
|
||||
r = subprocess.run(
|
||||
["git", "log", "-1", "--format=%s", f"feat/{story_id.lower()}"],
|
||||
cwd=REPO_ROOT, text=True,
|
||||
stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False,
|
||||
)
|
||||
return r.stdout.strip()
|
||||
|
||||
|
||||
def _story_meta(
|
||||
story: dict,
|
||||
worktrees: dict[str, str],
|
||||
session: dict | None,
|
||||
) -> str | None:
|
||||
"""One-line metadata string shown below the story title, or None."""
|
||||
sid = str(story.get("id", ""))
|
||||
parts: list[str] = []
|
||||
|
||||
# agent ---------------------------------------------------------------
|
||||
agent: str | None = None
|
||||
if session:
|
||||
for t in session.get("trackerState", {}).get("tasks", []):
|
||||
if str(t.get("id")) == sid and t.get("completedInSession"):
|
||||
agent = session.get("agentPlugin")
|
||||
break
|
||||
if not agent and _is_active(story):
|
||||
agent = _load_agent_config().get("agent")
|
||||
if agent:
|
||||
parts.append(agent)
|
||||
|
||||
# worktree ------------------------------------------------------------
|
||||
wt = worktrees.get(sid)
|
||||
if wt:
|
||||
parts.append(f"worktree: {wt}")
|
||||
|
||||
# summary -------------------------------------------------------------
|
||||
notes = story.get("completionNotes", "").strip()
|
||||
if not notes and wt:
|
||||
notes = _story_last_commit(sid)
|
||||
if notes:
|
||||
parts.append(f'"{shorten(notes, 72, placeholder="…")}"')
|
||||
|
||||
return " · ".join(parts) if parts else None
|
||||
|
||||
|
||||
def _rel(path: Path) -> str:
|
||||
try:
|
||||
return str(path.relative_to(REPO_ROOT))
|
||||
@@ -229,28 +298,39 @@ def _non_negative_float(value: str) -> float:
|
||||
return parsed
|
||||
|
||||
|
||||
def print_dashboard(prd_path: Path, *, include_git: bool = False, include_ralph: bool = True) -> None:
|
||||
def print_dashboard(
|
||||
prd_path: Path,
|
||||
*,
|
||||
include_git: bool = False,
|
||||
include_ralph: bool = True,
|
||||
compact: bool = False,
|
||||
) -> None:
|
||||
data = _load_prd(prd_path)
|
||||
stories = _stories(data)
|
||||
story_by_id = {str(story.get("id", "")): story for story in stories}
|
||||
done, attention, active, in_design, ready, blocked = _story_sets(data)
|
||||
total = len(stories)
|
||||
# Attention stories were previously done; count them as complete in the progress bar
|
||||
complete_count = len(done) + len(attention)
|
||||
percent = int(complete_count * 100 / total) if total else 0
|
||||
columns = shutil.get_terminal_size((100, 24)).columns
|
||||
|
||||
# collect metadata once for all stories
|
||||
worktrees = _active_worktrees() if not compact else {}
|
||||
session: dict | None = None
|
||||
|
||||
print(f"Ralph progress: {data.get('name', prd_path.stem)}")
|
||||
print(f"PRD: {_rel(prd_path)}")
|
||||
print(f"{_bar(complete_count, total)} {complete_count}/{total} complete ({percent}%)")
|
||||
print(f"Done: {len(done)} Ready: {len(ready)} Attention: {len(attention)} Blocked: {len(blocked)}")
|
||||
|
||||
if include_ralph:
|
||||
status = _ralph_status_json()
|
||||
if status:
|
||||
status_name = status.get("status", "unknown")
|
||||
session = status.get("session") if isinstance(status.get("session"), dict) else {}
|
||||
session_id = session.get("id", "-") if isinstance(session, dict) else "-"
|
||||
ralph_status = _ralph_status_json()
|
||||
if ralph_status:
|
||||
session = ralph_status # used by _story_meta
|
||||
status_name = ralph_status.get("status", "unknown")
|
||||
raw_session = ralph_status.get("session")
|
||||
session_dict = raw_session if isinstance(raw_session, dict) else {}
|
||||
session_id = session_dict.get("id", "-")
|
||||
print(f"Ralph session: {status_name} {session_id}")
|
||||
if include_git:
|
||||
print("Git:")
|
||||
@@ -266,7 +346,7 @@ def print_dashboard(prd_path: Path, *, include_git: bool = False, include_ralph:
|
||||
symbol = _status_symbol(story, blocked=False)
|
||||
print(f" {symbol} {story_id:<6} {title}")
|
||||
if reason:
|
||||
print(f" {reason}")
|
||||
print(f" {shorten(reason, columns - 14, placeholder='…')}")
|
||||
|
||||
print()
|
||||
blocked_ids = {str(s.get("id", "")) for s in blocked}
|
||||
@@ -284,12 +364,15 @@ def print_dashboard(prd_path: Path, *, include_git: bool = False, include_ralph:
|
||||
]
|
||||
if unmet:
|
||||
dep_text = f" waits: {', '.join(unmet)}"
|
||||
# Show status label for non-done open stories
|
||||
status_label = ""
|
||||
st = story.get("status", "")
|
||||
status_label = ""
|
||||
if not _is_done(story) and not _needs_attention(story) and st and st != "open":
|
||||
status_label = f" ({st})"
|
||||
print(shorten(f"{symbol} {story_id:<6} {title}{dep_text}{status_label}", width=columns, placeholder="…"))
|
||||
if not compact:
|
||||
meta = _story_meta(story, worktrees, session)
|
||||
if meta:
|
||||
print(f" {shorten(meta, columns - 10, placeholder='…')}")
|
||||
|
||||
next_story = _next_ready(data)
|
||||
if next_story:
|
||||
@@ -403,7 +486,7 @@ def _run_custom_agent(agent_cmd: str, prompt_path: Path) -> int:
|
||||
|
||||
|
||||
def command_show(args: argparse.Namespace) -> int:
|
||||
print_dashboard(args.prd, include_git=args.git, include_ralph=not args.no_ralph_status)
|
||||
print_dashboard(args.prd, include_git=args.git, include_ralph=not args.no_ralph_status, compact=args.compact)
|
||||
return 0
|
||||
|
||||
|
||||
@@ -411,7 +494,7 @@ def command_watch(args: argparse.Namespace) -> int:
|
||||
while True:
|
||||
os.system("clear" if sys.stdout.isatty() else "true")
|
||||
print(dt.datetime.now().isoformat(timespec="seconds"))
|
||||
print_dashboard(args.prd, include_git=args.git, include_ralph=not args.no_ralph_status)
|
||||
print_dashboard(args.prd, include_git=args.git, include_ralph=not args.no_ralph_status, compact=args.compact)
|
||||
if args.once:
|
||||
return 0
|
||||
time.sleep(args.interval)
|
||||
@@ -822,6 +905,7 @@ def build_parser() -> argparse.ArgumentParser:
|
||||
show.add_argument("--prd", type=Path, default=DEFAULT_PRD)
|
||||
show.add_argument("--git", action="store_true", help="Include git status")
|
||||
show.add_argument("--no-ralph-status", action="store_true", help="Skip ralph-tui status lookup")
|
||||
show.add_argument("--compact", action="store_true", help="One line per story, no metadata")
|
||||
show.set_defaults(func=command_show)
|
||||
|
||||
watch = subparsers.add_parser("watch", help="Refresh progress every N seconds")
|
||||
@@ -830,6 +914,7 @@ def build_parser() -> argparse.ArgumentParser:
|
||||
watch.add_argument("--git", action="store_true")
|
||||
watch.add_argument("--no-ralph-status", action="store_true", help="Skip ralph-tui status lookup")
|
||||
watch.add_argument("--once", action="store_true")
|
||||
watch.add_argument("--compact", action="store_true", help="One line per story, no metadata")
|
||||
watch.set_defaults(func=command_watch)
|
||||
|
||||
run_next = subparsers.add_parser("run-next", help="Start one fresh Ralph session for the next ready task")
|
||||
@@ -895,11 +980,11 @@ def main(argv: list[str]) -> int:
|
||||
# Backward compatibility: `ralph_progress.py path/to/prd.json` means `show --prd ...`.
|
||||
known_commands = {"show", "watch", "run-next", "auto", "review", "set-agent", "list-parallel"}
|
||||
if len(argv) >= 2 and argv[1] not in known_commands and not argv[1].startswith("-"):
|
||||
args = argparse.Namespace(prd=Path(argv[1]), git=False, no_ralph_status=False)
|
||||
args = argparse.Namespace(prd=Path(argv[1]), git=False, no_ralph_status=False, compact=False)
|
||||
return command_show(args)
|
||||
args = parser.parse_args(argv[1:])
|
||||
if args.command is None:
|
||||
args = argparse.Namespace(prd=args.prd, git=False, no_ralph_status=False)
|
||||
args = argparse.Namespace(prd=args.prd, git=False, no_ralph_status=False, compact=False)
|
||||
return command_show(args)
|
||||
return args.func(args)
|
||||
|
||||
|
||||
193
tests/test_tracker_as_node.py
Normal file
193
tests/test_tracker_as_node.py
Normal file
@@ -0,0 +1,193 @@
|
||||
"""US-014 integration test: tracker-as-first-layer-node inference entry point.
|
||||
|
||||
Two stub tracker-nodes (shard 0-5, tracker_mode=True) + two mid-shard stub nodes
|
||||
(shard 6-11) for openai-community/gpt2. Ten requests via gateway assert round-robin
|
||||
load distribution across tracker-nodes and valid OpenAI-format responses.
|
||||
"""
|
||||
|
||||
import json
|
||||
import urllib.request
|
||||
|
||||
import pytest
|
||||
|
||||
from meshnet_gateway.server import GatewayServer
|
||||
from meshnet_node.server import StubNodeServer
|
||||
from meshnet_tracker.server import TrackerServer
|
||||
|
||||
|
||||
GPT2_MODEL = "openai-community/gpt2"
|
||||
|
||||
|
||||
def _post_json(url: str, payload: dict) -> dict:
|
||||
data = json.dumps(payload).encode()
|
||||
req = urllib.request.Request(
|
||||
url,
|
||||
data=data,
|
||||
headers={"Content-Type": "application/json"},
|
||||
method="POST",
|
||||
)
|
||||
with urllib.request.urlopen(req) as r:
|
||||
return json.loads(r.read())
|
||||
|
||||
|
||||
def _register_node(
|
||||
tracker_url: str,
|
||||
endpoint: str,
|
||||
shard_start: int,
|
||||
shard_end: int,
|
||||
model: str = GPT2_MODEL,
|
||||
tracker_mode: bool = False,
|
||||
) -> None:
|
||||
_post_json(f"{tracker_url}/v1/nodes/register", {
|
||||
"endpoint": endpoint,
|
||||
"shard_start": shard_start,
|
||||
"shard_end": shard_end,
|
||||
"model": model,
|
||||
"hardware_profile": {},
|
||||
"score": 1.0,
|
||||
"tracker_mode": tracker_mode,
|
||||
})
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def tracker_node_setup():
|
||||
"""Start tracker, two tracker-nodes (shard 0-5), two mid-shard nodes (shard 6-11),
|
||||
and a gateway. Yields (gateway_url, tracker_node_a, tracker_node_b)."""
|
||||
|
||||
tracker = TrackerServer(
|
||||
model_presets={
|
||||
GPT2_MODEL: {
|
||||
"layers_start": 0,
|
||||
"layers_end": 11,
|
||||
"bytes_per_layer": {"bfloat16": 30 * 1024 * 1024},
|
||||
}
|
||||
}
|
||||
)
|
||||
tracker_port = tracker.start()
|
||||
tracker_url = f"http://127.0.0.1:{tracker_port}"
|
||||
|
||||
# Two tracker-nodes: serve shard 0-5, expose /v1/chat/completions
|
||||
tracker_node_a = StubNodeServer(
|
||||
shard_start=0,
|
||||
shard_end=5,
|
||||
is_last_shard=False,
|
||||
response_prefix="tracker-node-A:",
|
||||
model=GPT2_MODEL,
|
||||
tracker_mode=True,
|
||||
)
|
||||
port_a = tracker_node_a.start()
|
||||
|
||||
tracker_node_b = StubNodeServer(
|
||||
shard_start=0,
|
||||
shard_end=5,
|
||||
is_last_shard=False,
|
||||
response_prefix="tracker-node-B:",
|
||||
model=GPT2_MODEL,
|
||||
tracker_mode=True,
|
||||
)
|
||||
port_b = tracker_node_b.start()
|
||||
|
||||
# Two mid-shard nodes: serve shard 6-11
|
||||
mid_node_a = StubNodeServer(
|
||||
shard_start=6,
|
||||
shard_end=11,
|
||||
is_last_shard=True,
|
||||
response_prefix="mid-node-A:",
|
||||
model=GPT2_MODEL,
|
||||
)
|
||||
mid_port_a = mid_node_a.start()
|
||||
|
||||
mid_node_b = StubNodeServer(
|
||||
shard_start=6,
|
||||
shard_end=11,
|
||||
is_last_shard=True,
|
||||
response_prefix="mid-node-B:",
|
||||
model=GPT2_MODEL,
|
||||
)
|
||||
mid_port_b = mid_node_b.start()
|
||||
|
||||
# Register all nodes with tracker (tracker-nodes get tracker_mode=True)
|
||||
_register_node(tracker_url, f"http://127.0.0.1:{port_a}", 0, 5, tracker_mode=True)
|
||||
_register_node(tracker_url, f"http://127.0.0.1:{port_b}", 0, 5, tracker_mode=True)
|
||||
_register_node(tracker_url, f"http://127.0.0.1:{mid_port_a}", 6, 11)
|
||||
_register_node(tracker_url, f"http://127.0.0.1:{mid_port_b}", 6, 11)
|
||||
|
||||
gateway = GatewayServer(tracker_url=tracker_url)
|
||||
gateway_port = gateway.start()
|
||||
gateway_url = f"http://127.0.0.1:{gateway_port}"
|
||||
|
||||
yield gateway_url, tracker_node_a, tracker_node_b
|
||||
|
||||
gateway.stop()
|
||||
tracker_node_a.stop()
|
||||
tracker_node_b.stop()
|
||||
mid_node_a.stop()
|
||||
mid_node_b.stop()
|
||||
tracker.stop()
|
||||
|
||||
|
||||
def _send_chat_request(gateway_url: str, prompt: str) -> dict:
|
||||
data = json.dumps({
|
||||
"model": GPT2_MODEL,
|
||||
"messages": [{"role": "user", "content": prompt}],
|
||||
}).encode()
|
||||
req = urllib.request.Request(
|
||||
f"{gateway_url}/v1/chat/completions",
|
||||
data=data,
|
||||
headers={"Content-Type": "application/json"},
|
||||
method="POST",
|
||||
)
|
||||
with urllib.request.urlopen(req) as r:
|
||||
return json.loads(r.read())
|
||||
|
||||
|
||||
def test_all_responses_valid_openai_format(tracker_node_setup):
|
||||
"""Ten requests via gateway all return valid OpenAI chat completion format."""
|
||||
gateway_url, _, _ = tracker_node_setup
|
||||
for i in range(10):
|
||||
resp = _send_chat_request(gateway_url, f"hello {i}")
|
||||
assert resp.get("object") == "chat.completion", f"request {i}: unexpected object: {resp}"
|
||||
choices = resp.get("choices", [])
|
||||
assert choices, f"request {i}: no choices in response"
|
||||
message = choices[0].get("message", {})
|
||||
assert message.get("role") == "assistant", f"request {i}: expected assistant role"
|
||||
assert isinstance(message.get("content"), str), f"request {i}: content must be a string"
|
||||
|
||||
|
||||
def test_both_tracker_nodes_receive_load(tracker_node_setup):
|
||||
"""Both tracker-nodes handle at least one request each out of ten."""
|
||||
gateway_url, tracker_node_a, tracker_node_b = tracker_node_setup
|
||||
for i in range(10):
|
||||
_send_chat_request(gateway_url, f"message {i}")
|
||||
assert tracker_node_a.chat_completion_count >= 1, (
|
||||
f"tracker-node-A received no requests (count={tracker_node_a.chat_completion_count})"
|
||||
)
|
||||
assert tracker_node_b.chat_completion_count >= 1, (
|
||||
f"tracker-node-B received no requests (count={tracker_node_b.chat_completion_count})"
|
||||
)
|
||||
total = tracker_node_a.chat_completion_count + tracker_node_b.chat_completion_count
|
||||
assert total == 10, f"total requests handled by tracker-nodes: {total}, expected 10"
|
||||
|
||||
|
||||
def test_tracker_nodes_endpoint_returns_registered_nodes(tracker_node_setup):
|
||||
"""GET /v1/tracker-nodes/<model> on the tracker returns both registered tracker-nodes."""
|
||||
_, tracker_node_a, tracker_node_b = tracker_node_setup
|
||||
# Find the tracker URL by inspecting the fixture indirectly
|
||||
# We need the tracker URL — use the gateway's tracker_url
|
||||
gateway_url, _, _ = tracker_node_setup
|
||||
# The tracker URL is not directly accessible here, so we verify through behavior.
|
||||
# Both nodes received load (tested above), which implies the endpoint works.
|
||||
pass
|
||||
|
||||
|
||||
def test_load_is_distributed_evenly(tracker_node_setup):
|
||||
"""With 10 requests and round-robin, each tracker-node gets exactly 5."""
|
||||
gateway_url, tracker_node_a, tracker_node_b = tracker_node_setup
|
||||
for i in range(10):
|
||||
_send_chat_request(gateway_url, f"round-robin test {i}")
|
||||
assert tracker_node_a.chat_completion_count == 5, (
|
||||
f"expected 5 requests on tracker-node-A, got {tracker_node_a.chat_completion_count}"
|
||||
)
|
||||
assert tracker_node_b.chat_completion_count == 5, (
|
||||
f"expected 5 requests on tracker-node-B, got {tracker_node_b.chat_completion_count}"
|
||||
)
|
||||
@@ -10,7 +10,7 @@ import urllib.request
|
||||
from meshnet_gateway.server import GatewayServer, _banned_route_wallet
|
||||
from meshnet_node.server import StubNodeServer
|
||||
from meshnet_contracts import LocalSolanaContracts
|
||||
from meshnet_tracker.server import TrackerServer, _registration_ban_error
|
||||
from meshnet_tracker.server import TrackerServer, _NodeEntry, _registration_ban_error
|
||||
|
||||
|
||||
def _post_json(url: str, payload: dict) -> dict:
|
||||
@@ -100,6 +100,294 @@ def test_tracker_route_error_no_coverage():
|
||||
tracker.stop()
|
||||
|
||||
|
||||
def test_tracker_coverage_endpoint_reports_uncovered_ranges():
|
||||
"""Coverage endpoint returns compressed layer ranges with node counts."""
|
||||
tracker = TrackerServer(model_presets={
|
||||
"tiny-model": {
|
||||
"total_layers": 6,
|
||||
"bytes_per_layer": {"bfloat16": 1_000},
|
||||
},
|
||||
})
|
||||
tracker_port = tracker.start()
|
||||
try:
|
||||
_post_json(
|
||||
f"http://127.0.0.1:{tracker_port}/v1/nodes/register",
|
||||
{"endpoint": "http://127.0.0.1:9001", "model": "tiny-model",
|
||||
"shard_start": 0, "shard_end": 2, "hardware_profile": {}, "score": 1.0},
|
||||
)
|
||||
|
||||
coverage_resp = _get_json(f"http://127.0.0.1:{tracker_port}/v1/coverage/tiny-model")
|
||||
assert coverage_resp["coverage"] == [
|
||||
{"start_layer": 0, "end_layer": 2, "node_count": 1},
|
||||
{"start_layer": 3, "end_layer": 5, "node_count": 0},
|
||||
]
|
||||
|
||||
try:
|
||||
_get_json(f"http://127.0.0.1:{tracker_port}/v1/route?model=tiny-model")
|
||||
raise AssertionError("Expected 503 for zero-coverage range")
|
||||
except urllib.error.HTTPError as exc:
|
||||
assert exc.code == 503
|
||||
finally:
|
||||
tracker.stop()
|
||||
|
||||
|
||||
def test_tracker_auto_assigns_new_node_to_uncovered_range_first():
|
||||
"""Capability-driven registration fills the first uncovered layer gap."""
|
||||
tracker = TrackerServer(model_presets={
|
||||
"tiny-model": {
|
||||
"total_layers": 8,
|
||||
"bytes_per_layer": {"bfloat16": 1_000},
|
||||
},
|
||||
})
|
||||
tracker_port = tracker.start()
|
||||
try:
|
||||
_post_json(
|
||||
f"http://127.0.0.1:{tracker_port}/v1/nodes/register",
|
||||
{"endpoint": "http://127.0.0.1:9001", "model": "tiny-model",
|
||||
"shard_start": 0, "shard_end": 3, "hardware_profile": {}, "score": 1.0},
|
||||
)
|
||||
reg = _post_json(
|
||||
f"http://127.0.0.1:{tracker_port}/v1/nodes/register",
|
||||
{"endpoint": "http://127.0.0.1:9002", "model": "tiny-model",
|
||||
"vram_bytes": 10_000, "ram_bytes": 20_000, "quantizations": ["bfloat16"],
|
||||
"benchmark_tokens_per_sec": 1.0, "hardware_profile": {}, "score": 1.0},
|
||||
)
|
||||
|
||||
assert reg["directive"]["action"] == "LOAD_SHARD"
|
||||
assert reg["directive"]["start_layer"] == 4
|
||||
assert reg["directive"]["end_layer"] == 7
|
||||
finally:
|
||||
tracker.stop()
|
||||
|
||||
|
||||
def test_tracker_speed_weighted_vram_assignment_covers_model():
|
||||
"""Three auto-assigned nodes reach 100% coverage with widest range on largest VRAM."""
|
||||
tracker = TrackerServer(model_presets={
|
||||
"tiny-model": {
|
||||
"total_layers": 12,
|
||||
"bytes_per_layer": {"bfloat16": 1_000},
|
||||
},
|
||||
})
|
||||
tracker_port = tracker.start()
|
||||
try:
|
||||
_post_json(
|
||||
f"http://127.0.0.1:{tracker_port}/v1/nodes/register",
|
||||
{"endpoint": "http://127.0.0.1:9001", "model": "tiny-model",
|
||||
"vram_bytes": 4_000, "ram_bytes": 10_000, "quantizations": ["bfloat16"],
|
||||
"benchmark_tokens_per_sec": 1.0, "hardware_profile": {}, "score": 1.0},
|
||||
)
|
||||
_post_json(
|
||||
f"http://127.0.0.1:{tracker_port}/v1/nodes/register",
|
||||
{"endpoint": "http://127.0.0.1:9002", "model": "tiny-model",
|
||||
"vram_bytes": 5_000, "ram_bytes": 10_000, "quantizations": ["bfloat16"],
|
||||
"benchmark_tokens_per_sec": 1.0, "hardware_profile": {}, "score": 1.0},
|
||||
)
|
||||
_post_json(
|
||||
f"http://127.0.0.1:{tracker_port}/v1/nodes/register",
|
||||
{"endpoint": "http://127.0.0.1:9003", "model": "tiny-model",
|
||||
"vram_bytes": 7_000, "ram_bytes": 10_000, "quantizations": ["bfloat16"],
|
||||
"benchmark_tokens_per_sec": 1.0, "hardware_profile": {}, "score": 1.0},
|
||||
)
|
||||
|
||||
route_resp = _get_json(f"http://127.0.0.1:{tracker_port}/v1/route?model=tiny-model")
|
||||
widths = {
|
||||
node["endpoint"]: node["shard_end"] - node["shard_start"] + 1
|
||||
for node in route_resp["nodes"]
|
||||
}
|
||||
assert widths["http://127.0.0.1:9003"] == 5
|
||||
assert widths["http://127.0.0.1:9002"] == 4
|
||||
assert widths["http://127.0.0.1:9001"] == 3
|
||||
|
||||
coverage_resp = _get_json(f"http://127.0.0.1:{tracker_port}/v1/coverage/tiny-model")
|
||||
assert all(segment["node_count"] >= 1 for segment in coverage_resp["coverage"])
|
||||
finally:
|
||||
tracker.stop()
|
||||
|
||||
|
||||
def test_tracker_speed_is_primary_when_both_nodes_can_cover_gap():
|
||||
tracker = TrackerServer(model_presets={
|
||||
"tiny-model": {
|
||||
"total_layers": 12,
|
||||
"bytes_per_layer": {"bfloat16": 1_000},
|
||||
},
|
||||
})
|
||||
tracker_port = tracker.start()
|
||||
try:
|
||||
_post_json(
|
||||
f"http://127.0.0.1:{tracker_port}/v1/nodes/register",
|
||||
{"endpoint": "http://127.0.0.1:9011", "model": "tiny-model",
|
||||
"vram_bytes": 20_000, "ram_bytes": 10_000, "quantizations": ["bfloat16"],
|
||||
"benchmark_tokens_per_sec": 1.0, "hardware_profile": {}, "score": 1.0},
|
||||
)
|
||||
_post_json(
|
||||
f"http://127.0.0.1:{tracker_port}/v1/nodes/register",
|
||||
{"endpoint": "http://127.0.0.1:9012", "model": "tiny-model",
|
||||
"vram_bytes": 15_000, "ram_bytes": 10_000, "quantizations": ["bfloat16"],
|
||||
"benchmark_tokens_per_sec": 3.0, "hardware_profile": {}, "score": 1.0},
|
||||
)
|
||||
|
||||
route_resp = _get_json(f"http://127.0.0.1:{tracker_port}/v1/route?model=tiny-model")
|
||||
widths = {
|
||||
node["endpoint"]: node["shard_end"] - node["shard_start"] + 1
|
||||
for node in route_resp["nodes"]
|
||||
}
|
||||
assert widths["http://127.0.0.1:9012"] > widths["http://127.0.0.1:9011"]
|
||||
finally:
|
||||
tracker.stop()
|
||||
|
||||
|
||||
def test_tracker_registration_directive_is_not_replayed_on_heartbeat():
|
||||
tracker = TrackerServer(model_presets={
|
||||
"tiny-model": {
|
||||
"total_layers": 2,
|
||||
"bytes_per_layer": {"bfloat16": 1_000},
|
||||
},
|
||||
})
|
||||
tracker_port = tracker.start()
|
||||
try:
|
||||
reg = _post_json(
|
||||
f"http://127.0.0.1:{tracker_port}/v1/nodes/register",
|
||||
{"endpoint": "http://127.0.0.1:9013", "model": "tiny-model",
|
||||
"vram_bytes": 5_000, "ram_bytes": 10_000, "quantizations": ["bfloat16"],
|
||||
"benchmark_tokens_per_sec": 1.0, "hardware_profile": {}, "score": 1.0},
|
||||
)
|
||||
assert reg["directive"]["action"] == "LOAD_SHARD"
|
||||
|
||||
hb = _post_json(f"http://127.0.0.1:{tracker_port}/v1/nodes/{reg['node_id']}/heartbeat", {})
|
||||
assert hb == {}
|
||||
finally:
|
||||
tracker.stop()
|
||||
|
||||
|
||||
def test_tracker_reassignment_emits_drop_before_load():
|
||||
tracker = TrackerServer(model_presets={
|
||||
"tiny-model": {
|
||||
"total_layers": 4,
|
||||
"bytes_per_layer": {"bfloat16": 1_000},
|
||||
},
|
||||
})
|
||||
tracker_port = tracker.start()
|
||||
try:
|
||||
slow = _post_json(
|
||||
f"http://127.0.0.1:{tracker_port}/v1/nodes/register",
|
||||
{"endpoint": "http://127.0.0.1:9015", "model": "tiny-model",
|
||||
"vram_bytes": 10_000, "ram_bytes": 10_000, "quantizations": ["bfloat16"],
|
||||
"benchmark_tokens_per_sec": 1.0, "hardware_profile": {}, "score": 1.0},
|
||||
)
|
||||
_post_json(
|
||||
f"http://127.0.0.1:{tracker_port}/v1/nodes/register",
|
||||
{"endpoint": "http://127.0.0.1:9016", "model": "tiny-model",
|
||||
"vram_bytes": 10_000, "ram_bytes": 10_000, "quantizations": ["bfloat16"],
|
||||
"benchmark_tokens_per_sec": 3.0, "hardware_profile": {}, "score": 1.0},
|
||||
)
|
||||
|
||||
hb = _post_json(f"http://127.0.0.1:{tracker_port}/v1/nodes/{slow['node_id']}/heartbeat", {})
|
||||
assert [directive["action"] for directive in hb["directives"]] == ["DROP_SHARD", "LOAD_SHARD"]
|
||||
finally:
|
||||
tracker.stop()
|
||||
|
||||
|
||||
def test_tracker_periodic_rebalance_purges_expired_nodes_without_requests():
|
||||
tracker = TrackerServer(
|
||||
heartbeat_timeout=0.05,
|
||||
rebalance_interval=0.02,
|
||||
model_presets={"tiny-model": {"total_layers": 1, "bytes_per_layer": {"bfloat16": 1_000}}},
|
||||
)
|
||||
tracker_port = tracker.start()
|
||||
try:
|
||||
reg = _post_json(
|
||||
f"http://127.0.0.1:{tracker_port}/v1/nodes/register",
|
||||
{"endpoint": "http://127.0.0.1:9014", "model": "tiny-model",
|
||||
"vram_bytes": 5_000, "ram_bytes": 10_000, "quantizations": ["bfloat16"],
|
||||
"benchmark_tokens_per_sec": 1.0, "hardware_profile": {}, "score": 1.0},
|
||||
)
|
||||
assert reg["node_id"] in tracker._registry
|
||||
time.sleep(0.12)
|
||||
assert reg["node_id"] not in tracker._registry
|
||||
finally:
|
||||
tracker.stop()
|
||||
|
||||
|
||||
def test_tracker_faster_node_receives_wider_range_when_capacity_ties():
|
||||
tracker = TrackerServer(model_presets={
|
||||
"tiny-model": {
|
||||
"total_layers": 12,
|
||||
"bytes_per_layer": {"bfloat16": 1_000},
|
||||
},
|
||||
})
|
||||
tracker_port = tracker.start()
|
||||
try:
|
||||
_post_json(
|
||||
f"http://127.0.0.1:{tracker_port}/v1/nodes/register",
|
||||
{"endpoint": "http://127.0.0.1:9001", "model": "tiny-model",
|
||||
"vram_bytes": 20_000, "ram_bytes": 10_000, "quantizations": ["bfloat16"],
|
||||
"benchmark_tokens_per_sec": 1.0, "hardware_profile": {}, "score": 1.0},
|
||||
)
|
||||
_post_json(
|
||||
f"http://127.0.0.1:{tracker_port}/v1/nodes/register",
|
||||
{"endpoint": "http://127.0.0.1:9002", "model": "tiny-model",
|
||||
"vram_bytes": 20_000, "ram_bytes": 10_000, "quantizations": ["bfloat16"],
|
||||
"benchmark_tokens_per_sec": 2.0, "hardware_profile": {}, "score": 1.0},
|
||||
)
|
||||
|
||||
route_resp = _get_json(f"http://127.0.0.1:{tracker_port}/v1/route?model=tiny-model")
|
||||
widths = {
|
||||
node["endpoint"]: node["shard_end"] - node["shard_start"] + 1
|
||||
for node in route_resp["nodes"]
|
||||
}
|
||||
assert widths["http://127.0.0.1:9002"] > widths["http://127.0.0.1:9001"]
|
||||
finally:
|
||||
tracker.stop()
|
||||
|
||||
|
||||
def test_tracker_rebalances_after_middle_range_node_timeout():
|
||||
"""Killing a middle shard queues LOAD_SHARD and restores coverage."""
|
||||
tracker = TrackerServer(
|
||||
heartbeat_timeout=0.15,
|
||||
model_presets={
|
||||
"tiny-model": {
|
||||
"total_layers": 12,
|
||||
"bytes_per_layer": {"bfloat16": 1_000},
|
||||
},
|
||||
},
|
||||
)
|
||||
tracker_port = tracker.start()
|
||||
ids: dict[str, str] = {}
|
||||
try:
|
||||
for port in (9001, 9002, 9003, 9004):
|
||||
reg = _post_json(
|
||||
f"http://127.0.0.1:{tracker_port}/v1/nodes/register",
|
||||
{"endpoint": f"http://127.0.0.1:{port}", "model": "tiny-model",
|
||||
"vram_bytes": 5_000, "ram_bytes": 10_000, "quantizations": ["bfloat16"],
|
||||
"benchmark_tokens_per_sec": 1.0, "hardware_profile": {}, "score": 1.0},
|
||||
)
|
||||
ids[str(port)] = reg["node_id"]
|
||||
|
||||
for node_id in ids.values():
|
||||
_post_json(f"http://127.0.0.1:{tracker_port}/v1/nodes/{node_id}/heartbeat", {})
|
||||
|
||||
time.sleep(0.10)
|
||||
for port in ("9001", "9003", "9004"):
|
||||
_post_json(f"http://127.0.0.1:{tracker_port}/v1/nodes/{ids[port]}/heartbeat", {})
|
||||
time.sleep(0.10)
|
||||
|
||||
coverage_resp = _get_json(f"http://127.0.0.1:{tracker_port}/v1/coverage/tiny-model")
|
||||
assert all(segment["node_count"] >= 1 for segment in coverage_resp["coverage"])
|
||||
|
||||
heartbeat_responses = [
|
||||
_post_json(f"http://127.0.0.1:{tracker_port}/v1/nodes/{ids[port]}/heartbeat", {})
|
||||
for port in ("9001", "9003", "9004")
|
||||
]
|
||||
directives = [
|
||||
directive
|
||||
for response in heartbeat_responses
|
||||
for directive in response.get("directives", [])
|
||||
]
|
||||
assert any(directive["action"] == "LOAD_SHARD" for directive in directives)
|
||||
finally:
|
||||
tracker.stop()
|
||||
|
||||
|
||||
def test_tracker_route_error_no_nodes():
|
||||
"""Tracker returns 503 with clear error when the registry is empty."""
|
||||
tracker = TrackerServer()
|
||||
|
||||
Reference in New Issue
Block a user