From b5470347418643a5bc8795b59ccd3b978f15a1b0 Mon Sep 17 00:00:00 2001 From: Dobromir Popov Date: Mon, 6 Jul 2026 14:17:36 +0300 Subject: [PATCH] new tasks, devnet topup, cli, new model support --- QUICKSTART.md | 43 +++ accounts.sqlite | Bin 0 -> 12288 bytes billing.sqlite | Bin 12288 -> 12288 bytes .../meshnet-tracker-nobuild-stack.yml | 5 + docs/DEPLOY_PORTAINER.md | 22 +- docs/issues/36-relay-streamed-chat.md | 88 ++++++ docs/issues/37-relay-bridge-concurrency.md | 51 +++ docs/issues/38-tracker-seed-join.md | 62 ++++ docs/issues/39-caller-credit-account-keys.md | 60 ++++ docs/issues/40-devnet-dashboard-topup.md | 38 +++ docs/issues/41-account-wallet-keypair.md | 65 ++++ meshnet_registry.sqlite3 | Bin 0 -> 12288 bytes packages/node/meshnet_node/model_catalog.py | 34 +- packages/node/meshnet_node/relay_bridge.py | 126 +++++++- packages/node/meshnet_node/startup.py | 11 +- packages/relay/meshnet_relay/server.py | 33 +- packages/tracker/meshnet_tracker/accounts.py | 8 + packages/tracker/meshnet_tracker/billing.py | 23 ++ packages/tracker/meshnet_tracker/cli.py | 31 +- .../tracker/meshnet_tracker/dashboard.html | 11 +- packages/tracker/meshnet_tracker/server.py | 236 ++++++++++++-- tests/test_accounts.py | 97 +++++- tests/test_gossip_and_relay.py | 290 ++++++++++++++++++ tests/test_node_startup.py | 27 ++ 24 files changed, 1298 insertions(+), 63 deletions(-) create mode 100644 accounts.sqlite create mode 100644 docs/issues/36-relay-streamed-chat.md create mode 100644 docs/issues/37-relay-bridge-concurrency.md create mode 100644 docs/issues/38-tracker-seed-join.md create mode 100644 docs/issues/39-caller-credit-account-keys.md create mode 100644 docs/issues/40-devnet-dashboard-topup.md create mode 100644 docs/issues/41-account-wallet-keypair.md create mode 100644 meshnet_registry.sqlite3 diff --git a/QUICKSTART.md b/QUICKSTART.md index b02e71b..0c5beb3 100644 --- a/QUICKSTART.md +++ b/QUICKSTART.md @@ -54,6 +54,7 @@ on the host firewall if other machines will join: ```bash .venv/bin/meshnet-tracker start --host 0.0.0.0 --port 8080 +# --starting-credit 1 --devnet-topup 10 ``` Verify from the tracker host: @@ -514,6 +515,7 @@ Send inference through the tracker (which picks the head node and injects the ro ```bash curl -s https://ai.neuron.d-popov.com/v1/chat/completions \ -H "Content-Type: application/json" \ + -H "Authorization: Bearer sk-mesh-" \ -d '{ "model": "Qwen/Qwen2.5-0.5B-Instruct", "messages": [{"role": "user", "content": "What is 7 times 8?"}], @@ -529,6 +531,47 @@ curl -s http://localhost:7001/v1/chat/completions \ -d '{"model": "Qwen/Qwen2.5-0.5B-Instruct", "messages": [{"role": "user", "content": "Hi"}]}' ``` +## Accounts, API keys, and credit (billing-enabled trackers) + +Public trackers run with billing on: `/v1/chat/completions` requires a real +API key from a registered account. Unknown bearer strings get `401`; a key +with no balance gets `402 insufficient balance`. + +**Dashboard flow (easiest):** open `https:///dashboard`, register with +an email + password, then click **+ new key**. The key (`sk-mesh-…`) shows its +balance next to it. If the tracker was started with `--starting-credit`, your +first key arrives pre-funded (Caller Credit, once per account). If it was +started with `--devnet-topup`, every key row has a **+$N (devnet)** button to +refill during testing. + +**Curl flow:** + +```bash +# 1. Register (once) +curl -s https:///v1/auth/register \ + -H "Content-Type: application/json" \ + -d '{"email": "you@example.com", "password": "hunter22-or-better"}' +# → {"session_token": "...", ...} + +# 2. Create an API key (session token from step 1) +curl -s https:///v1/account/keys -X POST \ + -H "Authorization: Bearer " +# → {"api_key": "sk-mesh-...", "caller_credit_granted": true} + +# 3. Check balance / usage +curl -s https:///v1/account -H "Authorization: Bearer " + +# 4. (devnet trackers only) top up a key +curl -s https:///v1/account/topup -X POST \ + -H "Authorization: Bearer " \ + -H "Content-Type: application/json" \ + -d '{"api_key": "sk-mesh-..."}' +``` + +Operator side: both features default to 1 USDT (`--starting-credit` / +`--devnet-topup`). Set both to 0 on mainnet deployments — real deposits flow +through the on-chain USDT treasury watcher instead. + --- ## Step 1 — Start the tracker (Terminal 1) diff --git a/accounts.sqlite b/accounts.sqlite new file mode 100644 index 0000000000000000000000000000000000000000..dd139cfd9461c9ba1d69ac8cdee82ab3a4e159bf GIT binary patch literal 12288 zcmeI$O^X~w7y#g&kGRITjwm7KAT)yq7j`Dy^<8?&DkDf{vu2%uIW=#6Im7l$kJG&y z7xj=0UOadZ#6RG_@Mm}vMEni2m;T6z=O+;t!q0wjR*S`sN90q#qq4l?1`OK<$A06{^4M6 zG>Aug?+pj>`nmY+WkxnJ@o4bLDE{d1!~MO(d-1Koz1?^Yr+ER7&)hy3#kcPchr4lA z#(uzIyz1F`xB0@ggPle+nHl?e`C0D2<1;v}ie)=nzhIWGx9jiv7i-ULf3&Y$jb0gV z@4uY-N+>}f2n2y35Cnoi5C{T6;QuCYcD;QG9^Tnn*puSO9@*0#=E^8w3KBz!R7_YT zT`C>YLQ7)l7u|{Jz7uyZp3&WnyVdF3o@ZtLLUr%zpZcQ_eXHteYn0Nd1B8GmLx2hA zBvMLfMWt2CwE1K?f}cfs)SKFJ+ym(nrc%QRQNDuGlyWH%1ImTNCV# zb1MCeQZ6vYBuQwW2#%S+U+w+!-S2N^>6NqVt)-1#Sg4jZmZvU}f=dFJ$P_DLDELtw zuBYqN%I4x3Puo1odhTox#;lJMN@>Dso8P{C{oA+B-e`K8gRS#lN;z`^F@l7qgexVD z@Kns-KWiLOdAA1 zq?EXiF~^kDG{2&|KhTF)8<(=LZ}}zl>7=aeqNkGtQNo<}!2|&X36vPFw7|v~`$Xo# z2~WeUlIp(yxSZ@1CV9O$O`#UWbh7Bri+OQ!^AyHK(bvVa9{WDzxvhNE>^RSTdr{=e z1r268nbiSvD9iiB!enD8$MtB`(Imw(6~crKf8%RRiM7IB2I=8Pf7 zHPs4Sni`YVgO-rj*X&XO)V~F!b3PEX^`Tm*1QaudQ;q7_f4D7(OY!)=eE;W5{@vd0 zd=Pbh>>PJ~_LWeAKoAH5K_CbOfglhBf3sBu%qbq?E1IlU}O( zS9mI=2L%ryDBgPP#hd;Cp6x+!QlT`dU_qSI@IF77d7kIpnBSP+x;Wv!uFg%nh5F0N z6!=k?1VI4=;KS~8tzXUquwQZAVgdM5+ZWLJ!MXm~u3OycelOkMTymY}i>Izy5GxOi zFENZXjU`h?Vg(b)n3jl>FvdjL+nTqiy<0W!lz6^XoxR(8=(kL7!IvgdSX09oLClEA zK&W|QjF1dNu45`g*#)BEbD<}E?kj!N=nhOlagQ#Y9GUA)jI*TbL{I~HwI`E%m zj?;X8)m1A-FQ&ev5mJm|ZXm@GSAvJcHl?V5L|`nBjGqRd#4>+%Wj6bll@P7YYJ1py zScl_wh@BD6V~(uS?xeAq{bcK)^)1ja-`#N;$Lgum@YnLM`. Any key is accepted and starts with -the configured starter credit; calls return `402` after that balance is +requires `Authorization: Bearer ` — a real API key created through +an account (register on `/dashboard`, then "+ new key"); arbitrary bearer +strings are rejected with `401`. Calls return `402` once the key's balance is exhausted. Set `ENABLE_BILLING_DB=0` if existing unauthenticated clients must keep working during the first redeploy. +Credit variables (US-039/US-040): + +```text +STARTING_CREDIT=1 # one-time Caller Credit (USDT) on an account's first key +DEVNET_TOPUP=1 # dashboard "+$N (devnet)" faucet button; 0 disables +``` + +Both default to **1** (devnet-friendly alpha). On any deployment holding a +mainnet treasury set both to 0 — the faucet mints client balance for free. + Optional Solana treasury settlement variables: ```text @@ -94,7 +108,9 @@ Custom locations on the same proxy host: Leave sub-folder forwarding empty. -If WebSockets fail, Advanced: +**Required:** NPM custom locations do NOT inherit the proxy host's "Websockets +Support" toggle. Paste this into the **Advanced** box of *each* custom location +(`/ws` and `/rpc`), or every WebSocket handshake to the relay dies at nginx: ```nginx proxy_http_version 1.1; diff --git a/docs/issues/36-relay-streamed-chat.md b/docs/issues/36-relay-streamed-chat.md new file mode 100644 index 0000000..ea9efd2 --- /dev/null +++ b/docs/issues/36-relay-streamed-chat.md @@ -0,0 +1,88 @@ +# US-036 — Streamed chat completions over the relay RPC path + +Status: planned +Priority: Critical (blocks public friends-test deployment) +Stage: Designed + +## Context + +With the tracker deployed on a public VPS (`cloud.neuron.d-popov.com`), every node is +behind NAT, so **every** chat request is proxied tracker → relay → head node via +`_relay_http_request` (`packages/tracker/meshnet_tracker/server.py`). That function does a +single blocking `ws.recv()` and the node's `RelayHttpBridge._handle_request` does a single +`resp.read()`. Two consequences for `stream: true` requests: + +1. **No live streaming** — the client sees nothing until generation completes, then + receives the entire SSE body at once with a `Content-Length` header. +2. **Zero billing** — the tracker runs `_billable_non_stream_tokens(json.loads(body))` + on the buffered body; SSE text is not JSON, the parse fails, and the request is + billed/credited as 0 tokens. Off-LAN this silently zeroes out *all* streamed-request + accounting. + +Decision (grilled 2026-07-06): implement **true multi-frame streaming** over the relay +RPC WebSocket, scoped to the tracker → head-node leg. Rejected alternatives: +billing-only SSE parse (fragile heuristic, blank-screen UX stays) and forcing +`stream:false` over the relay (exact billing but still no live tokens for testers). +Streaming through the existing SSE accounting loop fixes both symptoms with one +mechanism. Inter-node `/forward` activation hops stay single-frame (ADR-0014) — they +are one-tensor-in/one-tensor-out and gain nothing from chunking. + +## Protocol + +A relayed response becomes a sequence of `relay-http-response` envelopes sharing one +`request_id`: + +```json +// first frame — status + headers, opens the stream +{"request_id": "", "status": 200, "headers": {"Content-Type": "text/event-stream"}, + "stream": true, "chunk": "data: {...}\n\n", "done": false} +// zero or more continuation frames +{"request_id": "", "stream": true, "chunk": "data: {...}\n\n", "done": false} +// terminal frame +{"request_id": "", "stream": true, "done": true} +``` + +**Backward compatibility:** a frame with no `stream` key is a complete single response +(today's format, still used for `/forward` hops, non-SSE responses, and older nodes). +The relay and tracker treat it as terminal. + +### Per component + +- **Node bridge** (`packages/node/meshnet_node/relay_bridge.py`): when the local + response `Content-Type` is `text/event-stream`, read line-by-line and emit chunk + frames as lines arrive; otherwise keep the existing single-frame path (including + `body_base64` for binary). Frame sends go through the bridge's WS send lock + (US-037) so frames from concurrent requests interleave whole, never torn. +- **Relay server** (`packages/relay/meshnet_relay/server.py`): `_handle_rpc` replaces + the single `asyncio.Future` in `_pending_rpc` with a per-request `asyncio.Queue`. + Frames arriving on the target peer's gossip connection are routed by `request_id` + and forwarded to the requester WS until `done` (or a terminal legacy frame). + Timeouts: keep the 310 s overall cap; add a 120 s per-frame idle timeout so a dead + node doesn't pin the queue. +- **Tracker** (`packages/tracker/meshnet_tracker/server.py`): `_relay_http_request` + grows a streaming mode — loop `ws.recv()`; on the first frame send status/headers to + the client; write each `chunk` to the client immediately and feed it through the + same SSE token-accounting used by the direct-proxy stream loop + (`reported_stream_tokens` / `_record_observed_throughput` / `_bill_completed`), so + relayed streams bill identically to direct streams. Non-stream frames keep the + current buffered handling. + +### Known limitation (accepted for alpha) + +If the client disconnects mid-stream, the relay drops undeliverable frames but the +node keeps generating until completion — wasted compute bounded by one generation. +Cancellation propagation is future work. + +## Acceptance criteria + +- `stream: true` chat request via relay delivers SSE chunks to the client + incrementally (test observes ≥2 distinct frame arrivals before `[DONE]`) +- Relayed streamed request records nonzero billed tokens and node work credit +- Non-streamed relayed requests and `/forward` binary hops behave exactly as before + (single frame, `body_base64` round-trip intact) +- Legacy single-frame response from an old node is accepted as terminal +- Idle stream (no frame for 120 s) returns 504 to the client and cleans up the + relay-side queue +- Extend `tests/test_gossip_and_relay.py` alongside + `test_relay_rpc_round_trips_http_request_to_peer` +- `python -m pytest` passes from repo root diff --git a/docs/issues/37-relay-bridge-concurrency.md b/docs/issues/37-relay-bridge-concurrency.md new file mode 100644 index 0000000..4f56d52 --- /dev/null +++ b/docs/issues/37-relay-bridge-concurrency.md @@ -0,0 +1,51 @@ +# US-037 — Concurrent request handling in the node relay bridge + +Status: planned +Priority: Critical (blocks public friends-test deployment) +Stage: Designed + +## Context + +`RelayHttpBridge._run` (`packages/node/meshnet_node/relay_bridge.py`) handles +`relay-http-request` envelopes serially inside its recv loop: `_handle_request` blocks +on `urllib.request.urlopen(..., timeout=300)` before the next envelope is read. + +Off-LAN this is a correctness bug, not just a throughput limit: a node can be the +**head** of one inference route and a **downstream hop** of another at the same time. +While the head request occupies the bridge (up to 300 s of generation), the other +route's per-token `/forward` calls sit unread in the WebSocket buffer — overlapping +routes through a shared node are effectively broken. + +Decision (grilled 2026-07-06): dispatch requests on a **bounded worker pool, default +8, configurable**. Rejected alternatives: unbounded thread-per-request (a public +deployment exposes volunteer machines to request stampedes) and reject-when-full +(bouncing `/forward` hops kills other routes' in-flight sessions; queueing beyond the +cap is today's behavior, just 8-wide). + +## Design + +- Recv loop only parses envelopes and submits to a + `ThreadPoolExecutor(max_workers=N)`; workers run `_handle_request` and send the + response frame(s). +- All sends on the single relay WebSocket go through a `threading.Lock`. With US-036 + a response may be multiple frames — the lock is held **per frame**, so streams from + concurrent requests interleave frame-atomically (receivers demux by `request_id`). +- `N` defaults to 8; configurable via `meshnet-node start --relay-concurrency N` + (env `MESHNET_RELAY_CONCURRENCY`). Requests beyond `N` queue in the executor. +- On reconnect (`_run`'s outer loop), in-flight workers from the dead connection may + still try to send; the send helper swallows failures on a closed socket, and the + relay side times the orphaned request out (US-036 idle timeout / existing 310 s cap). +- `stop()` shuts the executor down without waiting for stragglers (daemon threads, + same as today's bridge thread). + +## Acceptance criteria + +- While one relayed request is in flight (slow local handler), a second + `relay-http-request` to the same node completes without waiting for the first +- Responses are correctly matched by `request_id` when they complete out of order +- More than `N` simultaneous requests queue and all eventually complete; thread count + never exceeds `N` workers +- Bridge survives a relay reconnect with workers still in flight (no crash, no + deadlock; orphaned responses dropped) +- Extend `tests/test_gossip_and_relay.py` +- `python -m pytest` passes from repo root diff --git a/docs/issues/38-tracker-seed-join.md b/docs/issues/38-tracker-seed-join.md new file mode 100644 index 0000000..a7953fc --- /dev/null +++ b/docs/issues/38-tracker-seed-join.md @@ -0,0 +1,62 @@ +# US-038 — Tracker cluster join via a single seed peer + +Status: planned +Priority: High (blocks adding trackers without restarting the fleet) +Stage: Proposed + +## Context + +Today the tracker cluster is fully static. `RaftNode` and `NodeGossip` +(`packages/tracker/meshnet_tracker/raft.py`, `gossip.py`) take a peer list at +construction and never change it; `/v1/raft/status` does not expose membership. + +Consequences of starting a new tracker with only one existing peer in +`--cluster-peers`: + +- Existing trackers never learn about the newcomer — they keep heartbeating and + replicating to their original list only. +- Quorum math (`(len(self.peers) + 1) / 2`) differs per tracker, so vote counts + and commit decisions disagree — the cluster silently splits rather than erroring. + +Required behavior: a joining tracker is configured with **any one** live tracker +(a seed). It announces itself, the membership change replicates through the Raft +log, and every tracker — including the newcomer — converges on the same full peer +list. Removing the need to restart existing trackers when the hive grows. + +## Design + +1. **Expose membership.** `GET /v1/cluster/peers` returns + `{"self": , "peers": [, ...]}` (admin-safe: URLs only). +2. **Join handshake.** On startup, the joiner calls the seed's + `POST /v1/cluster/join` with its public self-url. The request is + hive-HMAC-signed (`MESHNET_HIVE_SECRET`, same fail-closed rule as gossip — + an unauthenticated join on a public tracker would let anyone enter the hive). + A non-leader seed forwards to the current leader (same pattern as + `/v1/nodes/register` forwarding). +3. **Membership through the log.** The leader appends a + `cluster-membership` entry (single-server change: add one peer per entry). + `RaftNode` gains `apply`-side handling that swaps `self.peers` — quorum is + recomputed from the applied membership, never from local config. Gossip peer + list updates from the same apply hook. +4. **Joiner bootstrap.** After a successful join response (which includes the + current peer list), the joiner sets its own peers and starts Raft as a + follower; it catches up via normal append-entries. +5. **Persistence.** Applied membership is written to the stats/state sqlite so a + restarted tracker rejoins with the last known list even if its seed is down. +6. **Config semantics.** `--cluster-peers` becomes "seed list": tried in order + until one join succeeds. A tracker with no seeds and no persisted membership + runs standalone (current single-tracker behavior unchanged). + +Out of scope: peer removal/eviction (leave via operator restart for now), +joint consensus for multi-server changes, automatic seed retry after startup. + +## Acceptance criteria + +- Start trackers A+B as a cluster; start C with only A as seed → within one + election timeout, A, B, and C all report the same 3-peer membership on + `GET /v1/cluster/peers`, and a value proposed on C commits on A and B +- Join without a valid hive signature is rejected with 403; join to a follower + is forwarded to the leader transparently +- Restarting C with its seed offline rejoins from persisted membership +- Standalone tracker (no seeds) behaves exactly as today +- `python -m pytest` passes from repo root diff --git a/docs/issues/39-caller-credit-account-keys.md b/docs/issues/39-caller-credit-account-keys.md new file mode 100644 index 0000000..2479f8f --- /dev/null +++ b/docs/issues/39-caller-credit-account-keys.md @@ -0,0 +1,60 @@ +# US-039 — Caller Credit granted once per account; chat requires account keys + +Status: planned +Priority: Critical (blocks friends-test inference — every request 402s) +Stage: Designed + +## Context + +`DEFAULT_STARTING_CREDIT = 0.0` and nothing can raise it: no CLI flag, no admin +credit endpoint. The only funding path is the on-chain USDT deposit watcher — +so on a fresh public tracker every inference request returns +`402 insufficient balance`, including keys created through the working +account flow (`/v1/auth/register` → `/v1/account/keys`, also in the dashboard). + +Separately, the billing gate accepts **any** bearer string (only revoked keys +are rejected), so wiring starter credit into `BillingLedger.ensure_client` +would let anyone mint unlimited free credit by inventing keys. + +Decision (grilled 2026-07-06): **Caller Credit is granted once per account**, +when the account creates its first API key; chat requests on an +accounts-enabled tracker must present a real active `sk-mesh-` key. +Rejected: credit-any-bearer (free-inference farming against volunteers' GPUs +on the open internet) and manual-admin-credit-only (operator toil, no +self-serve testing). + +## Design + +1. **`BillingLedger.grant_caller_credit(api_key, account_id, amount) -> bool`** + Appends a credit event with the deterministic id + `caller-credit-{account_id}` and note `caller-credit`. Already-seen id → + no-op returning False. The deterministic id makes the once-per-account rule + idempotent locally *and* across hive gossip (event dedupe is by id), so a + second key, a retried request, or a replicated event can never double-grant. +2. **`AccountStore.is_active_key(api_key) -> bool`.** The chat billing gate + (`_handle_proxy_chat`) rejects keys that are not active account keys with + `401 invalid_api_key` — but only when an accounts store is configured, so + embedded/test trackers without accounts keep today's any-key behavior. +3. **`--starting-credit N`** (USDT) on `meshnet-tracker start`. Default is the + single constant `DEFAULT_CALLER_CREDIT_USDT = 1.0` in `server.py` + (devnet-friendly alpha, revised 2026-07-06; set 0 on mainnet). Threaded + through `TrackerServer` → `_TrackerHTTPServer`; registration and + `_handle_account_key_create` call `grant_caller_credit` when billing is on + and the amount is positive. The ledger's own `_starting_credit` stays 0 — + implicit per-key grants in `ensure_client`/`charge_request` remain dead + code paths. +4. **Stack variable `STARTING_CREDIT`** in + `deploy/portainer/meshnet-tracker-nobuild-stack.yml` (default 1). +5. **Docs.** QUICKSTART gains "Get an API key" (register/login/create key via + dashboard or curl, what 402 means); DEPLOY_PORTAINER's stale "any key is + accepted and starts with the configured starter credit" paragraph replaced. + +## Acceptance criteria + +- Fresh account → first key → key has `--starting-credit` balance; chat succeeds +- Second key on the same account → no additional credit +- Revoke-and-recreate keys → still no additional credit (deterministic event id) +- Random bearer string on an accounts-enabled tracker → 401, never 402/free work +- Tracker without accounts store: gate behavior unchanged +- `--starting-credit 0` disables the grant entirely (mainnet posture) +- `python -m pytest` passes from repo root diff --git a/docs/issues/40-devnet-dashboard-topup.md b/docs/issues/40-devnet-dashboard-topup.md new file mode 100644 index 0000000..be1b62d --- /dev/null +++ b/docs/issues/40-devnet-dashboard-topup.md @@ -0,0 +1,38 @@ +# US-040 — Devnet top-up button on the dashboard + +Status: planned +Priority: High (friends-test convenience — refill without on-chain deposits) +Stage: Designed + +## Context + +Once Caller Credit (US-039) is spent, the only refill path is a real USDT +deposit through the treasury watcher — devnet Solana plumbing friends won't +have. For test deployments the dashboard needs a "add $10" style button. +This is a faucet: 0 disables it, and mainnet deployments must set 0. + +## Design + +1. **`--devnet-topup N`** (USDT per click) on `meshnet-tracker start`; stack + variable `DEVNET_TOPUP`. Default is the single constant + `DEFAULT_DEVNET_TOPUP_USDT = 1.0` in `server.py` (devnet-friendly alpha, + revised 2026-07-06 — the alpha's public trackers are devnet-only). +2. **`POST /v1/account/topup`** — session-authenticated (same as key + management). Body: `{"api_key": "sk-mesh-..."}`. Rules: + - 404 when the feature is disabled (flag absent/0) + - key must belong to the logged-in account (403 otherwise) + - credits exactly the configured amount (client cannot choose it), + note `devnet-topup`; returns the new balance +3. **Dashboard**: `/v1/account` response gains `topup_amount` (0 when + disabled); when positive, each key row renders a `+$N (devnet)` button + calling the endpoint. +4. **Abuse bound**: per-click amount is operator-set and requires a registered + session; acceptable for a friends-test faucet. Mainnet deployments simply + never set the flag. (Rate limiting is deliberately out of scope.) + +## Acceptance criteria + +- Flag off: endpoint 404s, dashboard shows no top-up button +- Flag on: logged-in user tops up own key, balance rises by exactly N +- Topping up another account's key → 403 +- `python -m pytest` passes from repo root diff --git a/docs/issues/41-account-wallet-keypair.md b/docs/issues/41-account-wallet-keypair.md new file mode 100644 index 0000000..8ebd026 --- /dev/null +++ b/docs/issues/41-account-wallet-keypair.md @@ -0,0 +1,65 @@ +# US-041 — Account wallet: browser-extension signing, in-browser generation, export-only + +Status: planned +Priority: Medium (not needed for devnet friends test; needed before mainnet) +Stage: Designed + +## Context + +Registration accepts an optional `wallet` field (a pubkey string) and the +dashboard never shows it. Users need a visible wallet address on the account +panel and a path to real Solana later, without the tracker ever holding user +private keys. + +Today wallets matter in two places: client deposit attribution +(`bind_wallet` → deposit watcher) and node payout identity (Pending Balance +keyed by wallet). Accounts themselves hold no keypair — and will stay that way. + +## Decision (2026-07-06) + +**Non-custodial.** The tracker stores public keys only; private keys never +leave the browser. + +1. **External wallets via browser extension.** The dashboard integrates the + standard Solana wallet-adapter flow (Phantom, Solflare, …): connect the + extension, prove ownership by signing a nonce, and the verified pubkey is + stored as the account wallet. All future transaction authorization + (deposits; later, anything requiring a user signature) happens in the + extension — the tracker only ever sees signatures and pubkeys. +2. **New wallets generated in-browser.** For users without an extension the + dashboard generates a keypair client-side (dashboard JS, `Ed25519` via + WebCrypto or bundled tweetnacl — CSP-safe, no CDN). The pubkey is uploaded; + the private key is shown once for export (base58, plaintext acceptable for + devnet) with a "save this now" warning, and optionally kept in + localStorage for convenience on devnet. +3. **No private-key import.** Users cannot paste an existing private key into + the dashboard — accepting foreign private keys is a liability the project + refuses. Users with existing wallets use the extension path (which imports + nothing; it signs in place). + +## Scope + +- Account panel shows the account wallet address (copy button) or the two + attach flows (connect extension / generate in browser) +- `POST /v1/account/wallet` — session-authenticated; body carries pubkey and, + for the extension flow, the signed-nonce ownership proof (reuse + `wallet_proof.py` if it fits); binds the wallet to the account's keys for + deposit attribution (existing `bind_wallet` path) +- Wallet is per-account; the tracker fans the binding out to the account's + API keys internally +- Mainnet cutover remains config-only (same boundary as Mock USDT, ADR-0015) + +Out of scope: password-wrapped keystore export, hardware wallets, changing +the node/miner wallet flow. + +## Acceptance criteria + +- Connect-extension flow stores a verified pubkey (rejects unsigned/mismatched + nonce proofs) +- Generate flow: pubkey lands on the account; private key is never sent to the + tracker (assert no such field in the request), export works +- No endpoint or UI accepts a private key +- Deposits to the shown address credit the account's keys via the existing + watcher +- Address visible on the account panel after either flow +- `python -m pytest` passes from repo root diff --git a/meshnet_registry.sqlite3 b/meshnet_registry.sqlite3 new file mode 100644 index 0000000000000000000000000000000000000000..b59a10b1fb1757756344dd7a61aab5708a55dfcc GIT binary patch literal 12288 zcmeI$J#W)M7y#h20~8R&h^6bv1C%C2vCsC|NJyk$BvjLnkTj%=b+#{+D>n|YlQcpM zVPxcQFfcPA{sm%#Uy%`C6cEaiOu+D-bYFh&J?kD%z86PcoO9ew2j?uu1l>ZCgdSmx z5GpNd(anogw3ilR(TVx0QbN^tw>Om;l9e5#>?ogBB|sbmKmY_l00ck)1V8`;KmY_l z00jQ6!1RW+u~jZh(}SD^2_Nv&ILikUpO1Ji&z7e06KaOzI zec8l^jc3p6jSk*(I~Ck#lO$z}$o@eS@3%a!g7XX)J9_xnlY0;4D_cip3B|pLUu3V6 z;xFE3!#thO{iQeh+R_8@Z+}&IZWdL(=^^D#(LezK5C8!X009sH0T2KI5C8!X0D(0G zrfvCh>uqP_EDi3nkViJ>glP^n%$jZM)QWVi%k02nhDlzlail)L>N;!H3Rd$;pNsWj zdfrcX&Lh9b6oD~Iin$2UGfJOkhaHWNj_rKdHBKn09}n8&i(R*^^+$W-*lLwSW0!)T&a)GIVOrPxe`#eYjqHS6TUpl&?ht1q46<1V8`;KmY_l h00ck)1V8`;K;VB6*etp}` literal 0 HcmV?d00001 diff --git a/packages/node/meshnet_node/model_catalog.py b/packages/node/meshnet_node/model_catalog.py index 51e2c91..2db2a21 100644 --- a/packages/node/meshnet_node/model_catalog.py +++ b/packages/node/meshnet_node/model_catalog.py @@ -159,6 +159,30 @@ CURATED_MODELS: list[ModelPreset] = [ ] +def layers_from_config(cfg) -> int | None: + """Extract the transformer layer count from a HuggingFace config object. + + Composite configs (vision-language, some MoE) nest the decoder settings in + ``text_config`` — e.g. Qwen3.5-MoE has no top-level ``num_hidden_layers``. + """ + candidates = [cfg] + get_text_config = getattr(cfg, "get_text_config", None) + if callable(get_text_config): + try: + candidates.append(get_text_config()) + except Exception: + pass + nested = getattr(cfg, "text_config", None) + if nested is not None: + candidates.append(nested) + for candidate in candidates: + for attr in ("num_hidden_layers", "num_layers", "n_layer", "n_layers"): + value = getattr(candidate, attr, None) + if value is not None: + return int(value) + return None + + def detect_num_layers(hf_repo: str) -> int | None: """Return num_hidden_layers from HuggingFace config.json (downloads ~1 KB only).""" # Check curated list first (no network call) @@ -168,7 +192,7 @@ def detect_num_layers(hf_repo: str) -> int | None: try: from transformers import AutoConfig # type: ignore[import] cfg = AutoConfig.from_pretrained(hf_repo) - return int(cfg.num_hidden_layers) + return layers_from_config(cfg) except Exception: return None @@ -195,6 +219,8 @@ def model_metadata_for( hf_repo, cache_dir=str(cache_dir) if cache_dir is not None else None, ) + # Composite configs (VLM/MoE) nest decoder fields in text_config. + text_cfg = getattr(cfg, "text_config", None) or cfg for attr, key in ( ("model_type", "architecture"), ("num_hidden_layers", "num_layers"), @@ -204,8 +230,14 @@ def model_metadata_for( ("max_position_embeddings", "context_length"), ): value = getattr(cfg, attr, None) + if value is None: + value = getattr(text_cfg, attr, None) if value is not None: metadata[key] = value + if "num_layers" not in metadata: + layers = layers_from_config(cfg) + if layers is not None: + metadata["num_layers"] = layers except Exception: pass return metadata diff --git a/packages/node/meshnet_node/relay_bridge.py b/packages/node/meshnet_node/relay_bridge.py index 4127160..721c7f1 100644 --- a/packages/node/meshnet_node/relay_bridge.py +++ b/packages/node/meshnet_node/relay_bridge.py @@ -5,14 +5,18 @@ from __future__ import annotations import base64 import json import logging +import os import threading import time import urllib.error import urllib.request +from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass log = logging.getLogger(__name__) +DEFAULT_MAX_CONCURRENCY = 8 + @dataclass(frozen=True) class RelayBridgeInfo: @@ -32,8 +36,23 @@ def _make_envelope(topic: str, payload: dict, peer_id: str) -> dict: } +def _max_concurrency_from_env() -> int: + try: + value = int(os.environ.get("MESHNET_RELAY_CONCURRENCY", DEFAULT_MAX_CONCURRENCY)) + except (TypeError, ValueError): + return DEFAULT_MAX_CONCURRENCY + return max(1, value) + + class RelayHttpBridge: - """Connect outbound to a relay and proxy relay HTTP requests to localhost.""" + """Connect outbound to a relay and proxy relay HTTP requests to localhost. + + Requests are dispatched on a bounded worker pool (US-037) so a node that is + the head of one route can still serve per-token ``/forward`` hops of another. + Streaming responses (``text/event-stream``) are forwarded as multiple chunk + frames sharing one ``request_id`` (US-036); everything else stays a single + frame for backward compatibility. + """ def __init__( self, @@ -42,15 +61,20 @@ class RelayHttpBridge: local_base_url: str, advertised_addr: str, reconnect_interval: float = 3.0, + max_concurrency: int | None = None, ) -> None: self.relay_url = relay_url.rstrip("/") self.peer_id = peer_id self.local_base_url = local_base_url.rstrip("/") self.advertised_addr = advertised_addr self.reconnect_interval = reconnect_interval + self.max_concurrency = max(1, max_concurrency) if max_concurrency else _max_concurrency_from_env() self._running = False self._thread: threading.Thread | None = None self._connected = threading.Event() + self._executor: ThreadPoolExecutor | None = None + self._send_lock = threading.Lock() + self._ws = None @property def relay_addr(self) -> str: @@ -61,6 +85,9 @@ class RelayHttpBridge: def start(self) -> RelayBridgeInfo: self._running = True + self._executor = ThreadPoolExecutor( + max_workers=self.max_concurrency, thread_name_prefix="relay-http-worker" + ) self._thread = threading.Thread(target=self._run, daemon=True, name="relay-http-bridge") self._thread.start() return RelayBridgeInfo(peer_id=self.peer_id, relay_addr=self.relay_addr) @@ -72,6 +99,8 @@ class RelayHttpBridge: self._running = False if self._thread: self._thread.join(timeout=3.0) + if self._executor is not None: + self._executor.shutdown(wait=False) def _run(self) -> None: import websockets.sync.client as wsc # type: ignore[import] @@ -79,6 +108,7 @@ class RelayHttpBridge: while self._running: try: with wsc.connect(self.relay_url, open_timeout=5) as ws: + self._ws = ws self._connected.set() ws.send(json.dumps(_make_envelope( "peer-register", @@ -99,19 +129,36 @@ class RelayHttpBridge: payload = envelope.get("payload", {}) if payload.get("target_peer") not in {None, self.peer_id}: continue - response = self._handle_request(payload) - ws.send(json.dumps(_make_envelope( - "relay-http-response", - response, - self.peer_id, - ))) + if self._executor is None: + break + self._executor.submit(self._process_request, payload) except Exception as exc: self._connected.clear() + self._ws = None if self._running: log.debug("relay bridge disconnected: %s", exc) time.sleep(self.reconnect_interval) + self._ws = None - def _handle_request(self, payload: dict) -> dict: + def _send_response_frame(self, payload: dict) -> bool: + """Send one relay-http-response frame; False if the socket is gone. + + The lock is held per frame so concurrent workers interleave whole + frames on the shared WebSocket, never torn ones. + """ + ws = self._ws + if ws is None: + return False + message = json.dumps(_make_envelope("relay-http-response", payload, self.peer_id)) + try: + with self._send_lock: + ws.send(message) + return True + except Exception as exc: + log.debug("relay bridge send failed (request orphaned): %s", exc) + return False + + def _process_request(self, payload: dict) -> None: request_id = str(payload.get("request_id") or "") method = str(payload.get("method") or "POST").upper() path = str(payload.get("path") or "/") @@ -130,10 +177,14 @@ class RelayHttpBridge: req = urllib.request.Request(url, data=data, headers=headers, method=method) try: with urllib.request.urlopen(req, timeout=300.0) as resp: - resp_bytes = resp.read() resp_headers = dict(resp.headers) + content_type = resp.headers.get("Content-Type", "") + if "text/event-stream" in content_type: + self._stream_response(request_id, resp, resp_headers) + return + resp_bytes = resp.read() # Forward all X-Meshnet-* headers so the caller can reconstruct the activation. - is_binary = "octet-stream" in resp.headers.get("Content-Type", "") + is_binary = "octet-stream" in content_type result: dict = { "request_id": request_id, "status": resp.status, @@ -143,21 +194,66 @@ class RelayHttpBridge: result["body_base64"] = base64.b64encode(resp_bytes).decode() else: result["body"] = resp_bytes.decode(errors="replace") - return result + self._send_response_frame(result) except urllib.error.HTTPError as exc: - return { + self._send_response_frame({ "request_id": request_id, "status": exc.code, "headers": {"Content-Type": exc.headers.get("Content-Type", "application/json")}, "body": exc.read().decode(errors="replace"), - } + }) except Exception as exc: - return { + self._send_response_frame({ "request_id": request_id, "status": 503, "headers": {"Content-Type": "application/json"}, "body": json.dumps({"error": f"relay bridge local request failed: {exc}"}), - } + }) + + def _stream_response(self, request_id: str, resp, resp_headers: dict) -> None: + """Forward an SSE response as chunk frames, one per complete SSE event. + + Frame order: header frame (status + headers), chunk frames, done frame. + A receiver that sees no ``stream`` key treats the frame as a complete + legacy response, so non-streaming peers are unaffected. + """ + sent = self._send_response_frame({ + "request_id": request_id, + "status": resp.status, + "headers": resp_headers, + "stream": True, + "done": False, + }) + if not sent: + return + event_lines: list[str] = [] + for raw_line in resp: + line = raw_line.decode(errors="replace") + event_lines.append(line) + if line.strip(): + continue + # Blank line terminates one SSE event — flush it as a frame. + if not self._send_response_frame({ + "request_id": request_id, + "stream": True, + "chunk": "".join(event_lines), + "done": False, + }): + return + event_lines = [] + if event_lines: + if not self._send_response_frame({ + "request_id": request_id, + "stream": True, + "chunk": "".join(event_lines), + "done": False, + }): + return + self._send_response_frame({ + "request_id": request_id, + "stream": True, + "done": True, + }) def peer_id_from_wallet(wallet_address: str) -> str: diff --git a/packages/node/meshnet_node/startup.py b/packages/node/meshnet_node/startup.py index 1aaf2ed..9a77394 100644 --- a/packages/node/meshnet_node/startup.py +++ b/packages/node/meshnet_node/startup.py @@ -734,11 +734,20 @@ def _detect_num_layers(model_id: str, cache_dir: Path | None = None) -> int | No """Fetch num_hidden_layers from HuggingFace model config (downloads ~1 KB config.json only).""" try: from transformers import AutoConfig # type: ignore[import] + + from .model_catalog import layers_from_config + cfg = AutoConfig.from_pretrained( model_id, cache_dir=str(cache_dir) if cache_dir is not None else None, ) - return int(cfg.num_hidden_layers) + layers = layers_from_config(cfg) + if layers is None: + print( + f" Warning: no layer count in {model_id} config " + "(checked top level and text_config)", flush=True, + ) + return layers except Exception as exc: print(f" Warning: could not read model config from HF: {exc}", flush=True) return None diff --git a/packages/relay/meshnet_relay/server.py b/packages/relay/meshnet_relay/server.py index fd73f76..c1f7ba0 100644 --- a/packages/relay/meshnet_relay/server.py +++ b/packages/relay/meshnet_relay/server.py @@ -57,7 +57,9 @@ class RelayServer: self._ready = threading.Event() self._running = False self._stop_event: asyncio.Event | None = None - self._pending_rpc: dict[str, asyncio.Future] = {} + # request_id → queue of relay-http-response frames (US-036: a streamed + # response is a sequence of frames; a frame without "stream" is terminal). + self._pending_rpc: dict[str, asyncio.Queue] = {} @property def registry(self) -> PeerRegistry: @@ -172,9 +174,9 @@ class RelayServer: if topic == "relay-http-response": payload = envelope.get("payload", {}) request_id = payload.get("request_id") - fut = self._pending_rpc.pop(request_id, None) - if fut is not None and not fut.done(): - fut.set_result(payload) + queue = self._pending_rpc.get(request_id) + if queue is not None: + queue.put_nowait(payload) continue # Fan out to all other registered peers @@ -240,8 +242,12 @@ class RelayServer: request_id = str(payload.get("request_id") or uuid.uuid4()) payload["request_id"] = request_id payload["target_peer"] = target_peer_id - fut = self._loop.create_future() if self._loop is not None else asyncio.get_running_loop().create_future() - self._pending_rpc[request_id] = fut + queue: asyncio.Queue = asyncio.Queue() + self._pending_rpc[request_id] = queue + overall_timeout = 310.0 + idle_timeout = 120.0 + loop = asyncio.get_running_loop() + deadline = loop.time() + overall_timeout try: await target.ws.send(json.dumps({ "topic": "relay-http-request", @@ -249,8 +255,19 @@ class RelayServer: "from_peer": "relay", "payload": payload, })) - response = await asyncio.wait_for(fut, timeout=310.0) - await ws_requester.send(json.dumps(response)) + # Forward frames until a terminal one: streamed responses (US-036) + # end with {"stream": true, "done": true}; a frame without "stream" + # is a complete legacy single response. + while True: + remaining = deadline - loop.time() + if remaining <= 0: + raise asyncio.TimeoutError + frame = await asyncio.wait_for( + queue.get(), timeout=min(idle_timeout, remaining) + ) + await ws_requester.send(json.dumps(frame)) + if not frame.get("stream") or frame.get("done"): + break except asyncio.TimeoutError: await ws_requester.send(json.dumps({ "request_id": request_id, diff --git a/packages/tracker/meshnet_tracker/accounts.py b/packages/tracker/meshnet_tracker/accounts.py index ef4a74a..626ce6f 100644 --- a/packages/tracker/meshnet_tracker/accounts.py +++ b/packages/tracker/meshnet_tracker/accounts.py @@ -176,6 +176,14 @@ class AccountStore: with self._lock: return api_key in self._revoked_keys + def is_active_key(self, api_key: str) -> bool: + with self._lock: + return api_key in self._active_keys + + def owner_of_key(self, api_key: str) -> str | None: + with self._lock: + return self._active_keys.get(api_key) + # ---- views ---- def _public_view(self, record: dict) -> dict: diff --git a/packages/tracker/meshnet_tracker/billing.py b/packages/tracker/meshnet_tracker/billing.py index 1a18087..0d99d48 100644 --- a/packages/tracker/meshnet_tracker/billing.py +++ b/packages/tracker/meshnet_tracker/billing.py @@ -87,6 +87,29 @@ class BillingLedger: def has_funds(self, api_key: str) -> bool: return self.ensure_client(api_key) > 0.0 + def grant_caller_credit(self, api_key: str, account_id: str, amount: float) -> bool: + """Grant the one-time Caller Credit for an account (US-039). + + The event id is derived from the account, not the key, so the grant is + idempotent per account — across retries, additional keys, and hive + gossip replication alike. Returns True only when credit was applied. + """ + if amount <= 0: + return False + event_id = f"caller-credit-{account_id}" + with self._lock: + if event_id in self._seen_event_ids: + return False + self._apply_locked({ + "id": event_id, + "type": "credit", + "api_key": api_key, + "amount": amount, + "ts": time.time(), + "note": "caller-credit", + }) + return True + def credit_client(self, api_key: str, amount: float, *, note: str = "deposit") -> float: if amount <= 0: raise ValueError("credit amount must be positive") diff --git a/packages/tracker/meshnet_tracker/cli.py b/packages/tracker/meshnet_tracker/cli.py index 3221479..88f8289 100644 --- a/packages/tracker/meshnet_tracker/cli.py +++ b/packages/tracker/meshnet_tracker/cli.py @@ -7,7 +7,12 @@ import time from .accounts import DEFAULT_ACCOUNTS_DB_PATH from .billing import DEFAULT_BILLING_DB_PATH from .hf_pricing import DEFAULT_HF_PRICING_LOG_DB_PATH -from .server import TrackerServer, derive_relay_url_from_public_tracker_url +from .server import ( + DEFAULT_CALLER_CREDIT_USDT, + DEFAULT_DEVNET_TOPUP_USDT, + TrackerServer, + derive_relay_url_from_public_tracker_url, +) DEFAULT_REGISTRY_DB_PATH = "meshnet_registry.sqlite3" @@ -66,6 +71,28 @@ def main() -> None: "token bound would cost more than this many USDT" ), ) + common.add_argument( + "--starting-credit", + type=float, + default=DEFAULT_CALLER_CREDIT_USDT, + metavar="USDT", + help=( + "One-time Caller Credit granted when an account creates its first " + f"API key (default: {DEFAULT_CALLER_CREDIT_USDT}; set 0 to require " + "deposits before inference)" + ), + ) + common.add_argument( + "--devnet-topup", + type=float, + default=DEFAULT_DEVNET_TOPUP_USDT, + metavar="USDT", + help=( + "Dashboard devnet top-up faucet: each click credits this many USDT " + f"to one of the account's keys (default: {DEFAULT_DEVNET_TOPUP_USDT}; " + "MUST be 0 on mainnet deployments)" + ), + ) common.add_argument( "--registry-db", default=DEFAULT_REGISTRY_DB_PATH, @@ -231,6 +258,8 @@ def main() -> None: enable_billing=not args.no_billing, billing_db=None if args.no_billing else args.billing_db, max_charge_per_request=args.max_charge_per_request, + starting_credit=args.starting_credit, + devnet_topup_amount=args.devnet_topup, contracts=contracts, accounts_db=None if args.no_accounts else args.accounts_db, treasury=treasury, diff --git a/packages/tracker/meshnet_tracker/dashboard.html b/packages/tracker/meshnet_tracker/dashboard.html index 63377cd..44b8883 100644 --- a/packages/tracker/meshnet_tracker/dashboard.html +++ b/packages/tracker/meshnet_tracker/dashboard.html @@ -297,6 +297,12 @@ async function revokeKey(key) { await renderAccountPanel(); } +async function topupKey(key) { + const r = await apiCall("/v1/account/topup", "POST", { api_key: key }); + if (!r.ok) alert(r.data.error || "top-up failed"); + await renderAccountPanel(); +} + async function renderAccountPanel() { const r = await apiCall("/v1/account"); if (r.status === 404) { // accounts disabled on this tracker @@ -305,7 +311,7 @@ async function renderAccountPanel() { return; } if (!r.ok) { setSession(null); renderAuthForms(); return; } - const { account, api_keys, balances, total_balance, usage } = r.data; + const { account, api_keys, balances, total_balance, usage, topup_amount } = r.data; const who = account.email || account.wallet || account.account_id; let html = `
${esc(who)} ${esc(account.role)} ` + @@ -319,6 +325,9 @@ async function renderAccountPanel() { for (const key of api_keys) { html += `
${esc(key)}` + ` (${usdt(balances[key] ?? 0)} USDT)` + + (topup_amount > 0 + ? ` ` + : "") + `
`; } } else { diff --git a/packages/tracker/meshnet_tracker/server.py b/packages/tracker/meshnet_tracker/server.py index 62888b7..3fff098 100644 --- a/packages/tracker/meshnet_tracker/server.py +++ b/packages/tracker/meshnet_tracker/server.py @@ -102,6 +102,11 @@ DEFAULT_VRAM_BYTES = 8 * 1024 * 1024 * 1024 DEFAULT_RAM_BYTES = 16 * 1024 * 1024 * 1024 DEFAULT_QUANTIZATIONS = ["bfloat16"] DEFAULT_BENCHMARK_TOKENS_PER_SEC = 1.0 +# US-039/US-040 — single source of truth for the credit defaults (referenced by +# TrackerServer, _TrackerHTTPServer, and the CLI). Alpha runs devnet-friendly; +# flip both to 0 before any deployment holding a mainnet treasury. +DEFAULT_CALLER_CREDIT_USDT = 1.0 +DEFAULT_DEVNET_TOPUP_USDT = 1.0 def _model_aliases(model: str | None) -> set[str]: @@ -909,18 +914,29 @@ def _coverage_gaps(coverage: list[dict]) -> list[tuple[int, int]]: ] -def _relay_http_request( +def _relay_http_request_frames( relay_addr: str, path: str, body: bytes, headers: dict[str, str], timeout: float = 310.0, -) -> dict | None: - """Send an HTTP-shaped request through a relay RPC WebSocket.""" + idle_timeout: float = 120.0, +): + """Send an HTTP-shaped request through a relay RPC WebSocket, yielding + response frames until a terminal one (US-036). + + A frame with ``stream: true`` is part of a chunked SSE response ending with + ``done: true``; a frame without ``stream`` is a complete single response. + Yields nothing when the relay is unreachable; stops silently on idle or + overall timeout (the caller bills whatever was observed). + """ try: import websockets.sync.client as wsc # type: ignore[import] - - request_id = str(uuid.uuid4()) + except Exception: + return + request_id = str(uuid.uuid4()) + deadline = time.monotonic() + timeout + try: with wsc.connect(relay_addr, open_timeout=10, close_timeout=5) as ws: ws.send(json.dumps({ "request_id": request_id, @@ -929,13 +945,63 @@ def _relay_http_request( "headers": headers, "body": body.decode(errors="replace"), })) - raw = ws.recv(timeout=timeout) - response = json.loads(raw) - if response.get("request_id") not in {None, request_id}: - return None - return response + while True: + remaining = deadline - time.monotonic() + if remaining <= 0: + return + raw = ws.recv(timeout=min(idle_timeout, remaining)) + frame = json.loads(raw) + if frame.get("request_id") not in {None, request_id}: + continue + yield frame + if not frame.get("stream") or frame.get("done"): + return except Exception: + return + + +def _relay_http_request( + relay_addr: str, + path: str, + body: bytes, + headers: dict[str, str], + timeout: float = 310.0, +) -> dict | None: + """Send an HTTP-shaped request through a relay and buffer the response. + + Streamed frame sequences are collapsed into one response dict whose body is + the concatenated SSE text — used by non-chat callers and kept for + backward compatibility; the chat proxy streams frames directly. + """ + frames = _relay_http_request_frames(relay_addr, path, body, headers, timeout=timeout) + first = next(frames, None) + if first is None: return None + if not first.get("stream"): + return first + chunks = [first.get("chunk") or ""] + for frame in frames: + chunks.append(frame.get("chunk") or "") + return { + "request_id": first.get("request_id"), + "status": first.get("status", 200), + "headers": first.get("headers") or {}, + "body": "".join(chunks), + } + + +def _stream_line_tokens(line: bytes) -> tuple[int, int | None]: + """Token accounting for one SSE line: (observed delta, reported total or None).""" + if not line.startswith(b"data:"): + return 0, None + payload = line[5:].strip() + if not payload or payload == b"[DONE]": + return 0, None + try: + chunk_payload = json.loads(payload) + except json.JSONDecodeError: + return 1, None + return _observed_stream_tokens(chunk_payload), _usage_total_tokens(chunk_payload) def _find_pinned_route( @@ -1507,6 +1573,8 @@ class _TrackerHTTPServer(socketserver.ThreadingMixIn, http.server.HTTPServer): validator_service_token: str | None = None, hive_secret: str | None = None, max_charge_per_request: float | None = None, + starting_credit: float = DEFAULT_CALLER_CREDIT_USDT, + devnet_topup_amount: float = DEFAULT_DEVNET_TOPUP_USDT, toploc_calibration: "ToplocCalibrationStore | None" = None, toploc_reference_node_url: str | None = None, toploc_calibration_gate_min_hardware_profiles: int = 1, @@ -1533,6 +1601,8 @@ class _TrackerHTTPServer(socketserver.ThreadingMixIn, http.server.HTTPServer): self.validator_service_token = validator_service_token self.hive_secret = hive_secret self.max_charge_per_request = max_charge_per_request + self.starting_credit = starting_credit + self.devnet_topup_amount = devnet_topup_amount self.toploc_calibration: ToplocCalibrationStore | None = toploc_calibration self.toploc_reference_node_url = ( toploc_reference_node_url.rstrip("/") if toploc_reference_node_url else None @@ -1672,6 +1742,9 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): if self.path == "/v1/account/keys/revoke": self._handle_account_key_revoke() return + if self.path == "/v1/account/topup": + self._handle_account_topup() + return if self.path == "/v1/accounts/gossip": self._handle_accounts_gossip() return @@ -2008,6 +2081,15 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): "code": "invalid_api_key", }}) return + # US-039: with accounts enabled, only real account keys may spend — + # arbitrary bearer strings must never become billable clients. + if server.accounts is not None and not server.accounts.is_active_key(api_key): + self._send_json(401, {"error": { + "message": "unknown API key: create one at /dashboard (register, then + new key)", + "type": "invalid_request_error", + "code": "invalid_api_key", + }}) + return if not server.billing.has_funds(api_key): self._send_json(402, {"error": { "message": "insufficient balance: deposit USDT to continue", @@ -2181,17 +2263,26 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): flush=True, ) started = time.monotonic() - relayed = _relay_http_request( + frames = _relay_http_request_frames( node.relay_addr, path="/v1/chat/completions", body=raw_body, headers=relay_headers, ) - elapsed = time.monotonic() - started - if relayed is not None: - self._send_relayed_response(relayed) - if int(relayed.get("status", 503)) < 400: - body_text = relayed.get("body") or "" + first = next(frames, None) + 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. + self._stream_relayed_frames( + first, frames, started, + model, route_model, route_nodes, api_key, node_work, + ) + return + if first is not None: + elapsed = time.monotonic() - started + self._send_relayed_response(first) + if int(first.get("status", 503)) < 400: + body_text = first.get("body") or "" try: tokens = _billable_non_stream_tokens(json.loads(body_text), body) except (json.JSONDecodeError, TypeError): @@ -2254,18 +2345,10 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): break self.wfile.write(line) self.wfile.flush() - if line.startswith(b"data:"): - payload = line[5:].strip() - if payload and payload != b"[DONE]": - try: - chunk_payload = json.loads(payload) - except json.JSONDecodeError: - observed_stream_tokens += 1 - continue - observed_stream_tokens += _observed_stream_tokens(chunk_payload) - found = _usage_total_tokens(chunk_payload) - if found is not None: - reported_stream_tokens = found + observed, reported = _stream_line_tokens(line) + observed_stream_tokens += observed + if reported is not None: + reported_stream_tokens = reported except BrokenPipeError: pass elapsed = time.monotonic() - started @@ -2410,6 +2493,49 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): except Exception as exc: print(f"[tracker] billing failed for model={model!r}: {exc}", flush=True) + def _stream_relayed_frames( + self, + first: dict, + frames, + started: float, + model: str, + route_model: str, + route_nodes: list, + api_key: str | None, + node_work: list, + ) -> None: + """Forward a streamed relay response (US-036) to the client as SSE, + billing with the same accounting as the direct stream path.""" + headers = first.get("headers") if isinstance(first.get("headers"), dict) else {} + self.send_response(int(first.get("status", 200))) + self.send_header("Content-Type", headers.get("Content-Type", "text/event-stream; charset=utf-8")) + self.send_header("Cache-Control", "no-cache") + self.end_headers() + reported_stream_tokens: int | None = None + observed_stream_tokens = 0 + client_gone = False + for frame in itertools.chain([first], frames): + chunk = frame.get("chunk") or "" + if not chunk: + continue + data = chunk.encode() + if not client_gone: + try: + self.wfile.write(data) + self.wfile.flush() + except BrokenPipeError: + # Keep draining frames — the nodes did the work; bill it. + client_gone = True + for line in data.splitlines(): + observed, reported = _stream_line_tokens(line) + observed_stream_tokens += observed + if reported is not None: + reported_stream_tokens = reported + elapsed = time.monotonic() - started + observed_tokens = _billable_stream_tokens(observed_stream_tokens, reported_stream_tokens) + self._record_observed_throughput(model, route_model, observed_tokens, elapsed, route_nodes) + self._bill_completed(api_key, model, observed_tokens, node_work) + def _send_relayed_response(self, response: dict) -> None: status = int(response.get("status", 503)) headers = response.get("headers") if isinstance(response.get("headers"), dict) else {} @@ -2909,6 +3035,12 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): api_key = accounts.create_api_key(account["account_id"]) server: _TrackerHTTPServer = self.server # type: ignore[assignment] if server.billing is not None: + # US-039: registration creates the account's first key, so Caller + # Credit lands here; the account-derived event id keeps later + # grants (e.g. via /v1/account/keys) no-ops. + server.billing.grant_caller_credit( + api_key, account["account_id"], server.starting_credit + ) server.billing.ensure_client(api_key) print( f"[tracker] account registered: {account.get('email') or account.get('wallet')} " @@ -2959,6 +3091,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): "balances": balances, "total_balance": sum(balances.values()), "usage": usage, + "topup_amount": server.devnet_topup_amount if server.billing is not None else 0.0, }) def _handle_account_key_create(self): @@ -2971,9 +3104,48 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): return api_key = accounts.create_api_key(account["account_id"]) server: _TrackerHTTPServer = self.server # type: ignore[assignment] + credited = False if server.billing is not None: + # US-039: Caller Credit lands with the account's first key; the + # account-derived event id keeps it once-per-account forever. + credited = server.billing.grant_caller_credit( + api_key, account["account_id"], server.starting_credit + ) server.billing.ensure_client(api_key) - self._send_json(200, {"api_key": api_key}) + self._send_json(200, {"api_key": api_key, "caller_credit_granted": credited}) + + def _handle_account_topup(self): + """Devnet faucet (US-040): credit the configured amount to one of the + logged-in account's keys. 404 unless the operator enabled it.""" + server: _TrackerHTTPServer = self.server # type: ignore[assignment] + accounts = self._require_accounts() + if accounts is None: + return + if server.billing is None or server.devnet_topup_amount <= 0: + self._send_json(404, {"error": "top-up is not enabled on this tracker"}) + return + account = self._session_account() + if account is None: + self._send_json(401, {"error": "login required"}) + return + body = self._read_json_body() + if body is None: + return + api_key = body.get("api_key") + if not api_key or not isinstance(api_key, str): + self._send_json(400, {"error": "api_key is required"}) + return + if accounts.owner_of_key(api_key) != account["account_id"]: + self._send_json(403, {"error": "key not found on this account"}) + return + balance = server.billing.credit_client( + api_key, server.devnet_topup_amount, note="devnet-topup" + ) + self._send_json(200, { + "api_key": api_key, + "credited": server.devnet_topup_amount, + "balance": balance, + }) def _handle_account_key_revoke(self): accounts = self._require_accounts() @@ -3851,6 +4023,8 @@ class TrackerServer: validator_service_token: str | None = None, hive_secret: str | None = None, max_charge_per_request: float | None = None, + starting_credit: float = DEFAULT_CALLER_CREDIT_USDT, + devnet_topup_amount: float = DEFAULT_DEVNET_TOPUP_USDT, toploc_calibration: ToplocCalibrationStore | None = None, toploc_calibration_db: str | None = None, toploc_reference_node_url: str | None = None, @@ -3923,6 +4097,8 @@ class TrackerServer: if max_charge_per_request is not None and max_charge_per_request <= 0.0: raise ValueError("max_charge_per_request must be positive") self._max_charge_per_request = max_charge_per_request + self._starting_credit = max(0.0, starting_credit) + self._devnet_topup_amount = max(0.0, devnet_topup_amount) self._validator_service_token = ( validator_service_token if validator_service_token is not None @@ -3976,6 +4152,8 @@ class TrackerServer: validator_service_token=self._validator_service_token, hive_secret=self._hive_secret, max_charge_per_request=self._max_charge_per_request, + starting_credit=self._starting_credit, + devnet_topup_amount=self._devnet_topup_amount, toploc_calibration=self._toploc_calibration, toploc_reference_node_url=self._toploc_reference_node_url, toploc_calibration_gate_min_hardware_profiles=self._toploc_calibration_gate_min_hardware_profiles, diff --git a/tests/test_accounts.py b/tests/test_accounts.py index 4da6a39..2c97140 100644 --- a/tests/test_accounts.py +++ b/tests/test_accounts.py @@ -124,9 +124,16 @@ def _call(url, method="GET", body=None, token=None): @pytest.fixture -def account_tracker(): - ledger = BillingLedger(starting_credit=0.0, default_price_per_1k=0.02) - tracker = TrackerServer(billing=ledger, accounts=AccountStore(), hive_secret=HIVE_SECRET) +def account_tracker(): + """Tracker with credit features pinned OFF (defaults are devnet-friendly 1.0).""" + ledger = BillingLedger(starting_credit=0.0, default_price_per_1k=0.02) + tracker = TrackerServer( + billing=ledger, + accounts=AccountStore(), + hive_secret=HIVE_SECRET, + starting_credit=0.0, + devnet_topup_amount=0.0, + ) port = tracker.start() yield f"http://127.0.0.1:{port}", ledger tracker.stop() @@ -145,7 +152,7 @@ def test_register_login_and_account_view(account_tracker): me = _call(f"{url}/v1/account", token=login["session_token"]) assert me["account"]["email"] == "admin@example.com" assert me["api_keys"] == [reg["api_key"]] - assert me["total_balance"] == pytest.approx(0.0) + assert me["total_balance"] == pytest.approx(0.0) assert me["usage"]["requests"] == 0 @@ -228,3 +235,85 @@ def test_accounts_endpoints_404_when_disabled(): assert exc_info.value.code == 404 finally: tracker.stop() + + +# ------------------------------------------- US-039/US-040: credit and top-up + + +@pytest.fixture +def funded_tracker(): + """Tracker with Caller Credit and the devnet top-up faucet enabled.""" + ledger = BillingLedger(starting_credit=0.0, default_price_per_1k=0.02) + tracker = TrackerServer( + billing=ledger, + accounts=AccountStore(), + hive_secret=HIVE_SECRET, + starting_credit=1.0, + devnet_topup_amount=10.0, + ) + port = tracker.start() + yield f"http://127.0.0.1:{port}", ledger + tracker.stop() + + +def test_caller_credit_granted_once_per_account(funded_tracker): + url, ledger = funded_tracker + reg = _call(f"{url}/v1/auth/register", "POST", + {"email": "c@example.com", "password": "secret-123"}) + token = reg["session_token"] + first_key = reg["api_key"] + assert ledger.get_client_balance(first_key) == pytest.approx(1.0) + + # A second key never re-grants — not even after revoking the first. + second = _call(f"{url}/v1/account/keys", "POST", {}, token=token) + assert second["caller_credit_granted"] is False + assert ledger.get_client_balance(second["api_key"]) == pytest.approx(0.0) + _call(f"{url}/v1/account/keys/revoke", "POST", {"api_key": first_key}, token=token) + third = _call(f"{url}/v1/account/keys", "POST", {}, token=token) + assert third["caller_credit_granted"] is False + assert ledger.get_client_balance(third["api_key"]) == pytest.approx(0.0) + + +def test_unknown_bearer_key_rejected_by_proxy(funded_tracker): + url, ledger = funded_tracker + with pytest.raises(urllib.error.HTTPError) as exc_info: + _call(f"{url}/v1/chat/completions", "POST", + {"model": "any", "messages": []}, token="sk-mesh-made-up-key") + assert exc_info.value.code == 401 + assert "unknown API key" in exc_info.value.read().decode() + # The invented key must not have become a billable client. + assert ledger.get_client_balance("sk-mesh-made-up-key") == pytest.approx(0.0) + + +def test_devnet_topup_credits_own_key_only(funded_tracker): + url, ledger = funded_tracker + owner = _call(f"{url}/v1/auth/register", "POST", + {"email": "own@example.com", "password": "secret-123"}) + other = _call(f"{url}/v1/auth/register", "POST", + {"email": "oth@example.com", "password": "secret-123"}) + + me = _call(f"{url}/v1/account", token=owner["session_token"]) + assert me["topup_amount"] == pytest.approx(10.0) + + result = _call(f"{url}/v1/account/topup", "POST", + {"api_key": owner["api_key"]}, token=owner["session_token"]) + assert result["credited"] == pytest.approx(10.0) + assert result["balance"] == pytest.approx(11.0) # 1.0 caller credit + 10.0 + + with pytest.raises(urllib.error.HTTPError) as exc_info: + _call(f"{url}/v1/account/topup", "POST", + {"api_key": owner["api_key"]}, token=other["session_token"]) + assert exc_info.value.code == 403 + assert ledger.get_client_balance(owner["api_key"]) == pytest.approx(11.0) + + +def test_topup_404_when_disabled(account_tracker): + url, _ = account_tracker + reg = _call(f"{url}/v1/auth/register", "POST", + {"email": "t@example.com", "password": "secret-123"}) + me = _call(f"{url}/v1/account", token=reg["session_token"]) + assert me["topup_amount"] == pytest.approx(0.0) + with pytest.raises(urllib.error.HTTPError) as exc_info: + _call(f"{url}/v1/account/topup", "POST", + {"api_key": reg["api_key"]}, token=reg["session_token"]) + assert exc_info.value.code == 404 diff --git a/tests/test_gossip_and_relay.py b/tests/test_gossip_and_relay.py index 5530a04..4dc62a9 100644 --- a/tests/test_gossip_and_relay.py +++ b/tests/test_gossip_and_relay.py @@ -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"} diff --git a/tests/test_node_startup.py b/tests/test_node_startup.py index acb7c25..d277325 100644 --- a/tests/test_node_startup.py +++ b/tests/test_node_startup.py @@ -1387,3 +1387,30 @@ def test_startup_cpu_fallback(tmp_path, monkeypatch): node.stop() finally: tracker.stop() + + +# --------------------------------------------------- layer detection (US: composite configs) + + +def test_layers_from_config_top_level(): + from meshnet_node.model_catalog import layers_from_config + + cfg = types.SimpleNamespace(num_hidden_layers=24) + assert layers_from_config(cfg) == 24 + + +def test_layers_from_config_nested_text_config(): + """VLM/MoE composites (e.g. Qwen3.5-MoE) keep the layer count in text_config.""" + from meshnet_node.model_catalog import layers_from_config + + cfg = types.SimpleNamespace(text_config=types.SimpleNamespace(num_hidden_layers=40)) + assert layers_from_config(cfg) == 40 + + +def test_layers_from_config_get_text_config_and_variants(): + from meshnet_node.model_catalog import layers_from_config + + inner = types.SimpleNamespace(n_layer=32) + cfg = types.SimpleNamespace(get_text_config=lambda: inner) + assert layers_from_config(cfg) == 32 + assert layers_from_config(types.SimpleNamespace()) is None