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

new tasks, devnet topup, cli, new model support

Browse Source
This commit is contained in:
Dobromir Popov
2026-07-06 14:17:36 +03:00
parent f841dfaeed
commit b547034741
24 changed files with 1298 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files

290
tests/test_gossip_and_relay.py
View File

@@ -525,3 +525,293 @@ def test_mdns_start_and_stop_without_zeroconf(monkeypatch):
disc = MdnsDiscovery(peer_id="x", port=8001)
disc.start() # should not raise
disc.stop() # should not raise
# ---------------------------------------------------------------------------
# US-036: streamed responses over relay RPC / US-037: bridge concurrency
# ---------------------------------------------------------------------------
def _start_local_http_server(handler_cls):
from http.server import ThreadingHTTPServer
server = ThreadingHTTPServer(("127.0.0.1", 0), handler_cls)
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()
return server, server.server_address[1]
def _run_streaming_peer(port: int, peer_id: str, ready: threading.Event):
"""Peer that answers one rpc request with a 4-frame streamed response."""
import websockets.sync.client as wsc # type: ignore[import]
with wsc.connect(f"ws://127.0.0.1:{port}/ws") as ws:
ws.send(json.dumps({
"topic": "peer-register",
"version": 1,
"from_peer": peer_id,
"msg_id": f"{peer_id}-reg",
"payload": {"peer_id": peer_id, "addr": ""},
}))
ws.recv()
ready.set()
envelope = json.loads(ws.recv(timeout=5))
rid = envelope["payload"]["request_id"]
def frame(payload):
ws.send(json.dumps({
"topic": "relay-http-response",
"version": 1,
"from_peer": peer_id,
"payload": payload,
}))
frame({"request_id": rid, "status": 200,
"headers": {"Content-Type": "text/event-stream"},
"stream": True, "done": False})
frame({"request_id": rid, "stream": True,
"chunk": 'data: {"choices": [{"delta": {"content": "hello"}}]}\n\n',
"done": False})
frame({"request_id": rid, "stream": True,
"chunk": "data: [DONE]\n\n", "done": False})
frame({"request_id": rid, "stream": True, "done": True})
def test_relay_rpc_forwards_streamed_frames_in_order():
"""A streamed response traverses the relay as multiple frames ending with done."""
import websockets.sync.client as wsc # type: ignore[import]
from meshnet_relay.server import RelayServer
relay = RelayServer(host="127.0.0.1", port=0)
port = relay.start()
ready = threading.Event()
peer_thread = threading.Thread(
target=_run_streaming_peer, args=(port, "stream_peer", ready), daemon=True
)
peer_thread.start()
assert ready.wait(timeout=5)
frames = []
with wsc.connect(f"ws://127.0.0.1:{port}/rpc/stream_peer") as ws:
ws.send(json.dumps({
"request_id": "req-stream-1",
"method": "POST",
"path": "/v1/chat/completions",
"headers": {},
"body": "{}",
}))
while True:
frames.append(json.loads(ws.recv(timeout=5)))
if not frames[-1].get("stream") or frames[-1].get("done"):
break
relay.stop()
peer_thread.join(timeout=3)
assert len(frames) == 4
assert frames[0]["status"] == 200 and frames[0]["stream"] is True
assert frames[1]["chunk"].startswith("data:")
assert frames[2]["chunk"] == "data: [DONE]\n\n"
assert frames[-1]["done"] is True
def test_tracker_relay_http_request_collapses_streamed_frames():
"""Buffered wrapper joins chunk frames into one response body."""
from meshnet_relay.server import RelayServer
from meshnet_tracker.server import _relay_http_request
relay = RelayServer(host="127.0.0.1", port=0)
port = relay.start()
ready = threading.Event()
peer_thread = threading.Thread(
target=_run_streaming_peer, args=(port, "buffer_peer", ready), daemon=True
)
peer_thread.start()
assert ready.wait(timeout=5)
response = _relay_http_request(
f"ws://127.0.0.1:{port}/rpc/buffer_peer",
path="/v1/chat/completions",
body=b"{}",
headers={},
)
relay.stop()
peer_thread.join(timeout=3)
assert response is not None
assert response["status"] == 200
assert '"content": "hello"' in response["body"]
assert "data: [DONE]" in response["body"]
def test_stream_line_tokens_accounting():
from meshnet_tracker.server import _stream_line_tokens
assert _stream_line_tokens(b": comment") == (0, None)
assert _stream_line_tokens(b"data: [DONE]") == (0, None)
observed, reported = _stream_line_tokens(
b'data: {"choices": [{"delta": {"content": "two words"}}]}'
)
assert observed == 2 and reported is None
observed, reported = _stream_line_tokens(b'data: {"usage": {"total_tokens": 42}}')
assert reported == 42
assert _stream_line_tokens(b"data: not-json") == (1, None)
def test_relay_bridge_streams_sse_as_chunk_frames():
"""Bridge forwards a local SSE response as header + chunk frames + done."""
from http.server import BaseHTTPRequestHandler
from meshnet_node.relay_bridge import RelayHttpBridge
class SSEHandler(BaseHTTPRequestHandler):
def do_POST(self):
self.rfile.read(int(self.headers.get("Content-Length", 0) or 0))
self.send_response(200)
self.send_header("Content-Type", "text/event-stream")
self.end_headers()
for i in range(3):
self.wfile.write(f'data: {{"i": {i}}}\n\n'.encode())
self.wfile.flush()
self.wfile.write(b"data: [DONE]\n\n")
def log_message(self, *args):
pass
server, port = _start_local_http_server(SSEHandler)
try:
bridge = RelayHttpBridge(
relay_url="ws://unused.example/ws",
peer_id="peer-sse",
local_base_url=f"http://127.0.0.1:{port}",
advertised_addr="",
)
frames = []
bridge._send_response_frame = lambda payload: (frames.append(payload), True)[1]
bridge._process_request({
"request_id": "req-sse-1",
"method": "POST",
"path": "/v1/chat/completions",
"headers": {"Content-Type": "application/json"},
"body": "{}",
})
finally:
server.shutdown()
assert frames[0]["stream"] is True and frames[0]["status"] == 200
chunk_frames = [f for f in frames if f.get("chunk")]
assert len(chunk_frames) >= 2
assert frames[-1] == {"request_id": "req-sse-1", "stream": True, "done": True}
assert "".join(f["chunk"] for f in chunk_frames).count("data:") == 4
def test_relay_bridge_non_stream_response_stays_single_frame():
from http.server import BaseHTTPRequestHandler
from meshnet_node.relay_bridge import RelayHttpBridge
class JSONHandler(BaseHTTPRequestHandler):
def do_POST(self):
self.rfile.read(int(self.headers.get("Content-Length", 0) or 0))
body = json.dumps({"ok": True}).encode()
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def log_message(self, *args):
pass
server, port = _start_local_http_server(JSONHandler)
try:
bridge = RelayHttpBridge(
relay_url="ws://unused.example/ws",
peer_id="peer-json",
local_base_url=f"http://127.0.0.1:{port}",
advertised_addr="",
)
frames = []
bridge._send_response_frame = lambda payload: (frames.append(payload), True)[1]
bridge._process_request({
"request_id": "req-json-1",
"method": "POST",
"path": "/v1/chat/completions",
"headers": {"Content-Type": "application/json"},
"body": "{}",
})
finally:
server.shutdown()
assert len(frames) == 1
assert "stream" not in frames[0]
assert json.loads(frames[0]["body"]) == {"ok": True}
def test_relay_bridge_serves_concurrent_requests():
"""US-037: a slow relayed request must not block a fast one."""
import websockets.sync.client as wsc # type: ignore[import]
from http.server import BaseHTTPRequestHandler
from meshnet_node.relay_bridge import RelayHttpBridge
from meshnet_relay.server import RelayServer
class Handler(BaseHTTPRequestHandler):
def do_POST(self):
self.rfile.read(int(self.headers.get("Content-Length", 0) or 0))
if self.path == "/slow":
time.sleep(2.0)
body = json.dumps({"path": self.path}).encode()
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def log_message(self, *args):
pass
http_server, http_port = _start_local_http_server(Handler)
relay = RelayServer(host="127.0.0.1", port=0)
relay_port = relay.start()
bridge = RelayHttpBridge(
relay_url=f"ws://127.0.0.1:{relay_port}/ws",
peer_id="conc_peer",
local_base_url=f"http://127.0.0.1:{http_port}",
advertised_addr="",
)
bridge.start()
results = {}
try:
assert bridge.wait_connected(timeout=5)
deadline = time.time() + 5
while time.time() < deadline and relay.registry.get("conc_peer") is None:
time.sleep(0.05)
assert relay.registry.get("conc_peer") is not None
def call(path, key):
with wsc.connect(f"ws://127.0.0.1:{relay_port}/rpc/conc_peer") as ws:
ws.send(json.dumps({
"request_id": f"req-{key}",
"method": "POST",
"path": path,
"headers": {"Content-Type": "application/json"},
"body": "{}",
}))
results[key] = (json.loads(ws.recv(timeout=10)), time.monotonic())
start = time.monotonic()
slow_thread = threading.Thread(target=call, args=("/slow", "slow"), daemon=True)
slow_thread.start()
time.sleep(0.3) # let the slow request reach the bridge first
call("/fast", "fast")
fast_elapsed = results["fast"][1] - start
slow_thread.join(timeout=10)
finally:
bridge.stop()
relay.stop()
http_server.shutdown()
assert results["fast"][0]["status"] == 200
assert results["slow"][0]["status"] == 200
# The fast request finished while the slow one was still in flight.
assert fast_elapsed < 1.5
# Responses matched by request_id despite out-of-order completion.
assert json.loads(results["fast"][0]["body"]) == {"path": "/fast"}
assert json.loads(results["slow"][0]["body"]) == {"path": "/slow"}