From f841dfaeed6e3e7d029611eecd8c93944f736eec Mon Sep 17 00:00:00 2001 From: Dobromir Popov Date: Mon, 6 Jul 2026 09:48:27 +0300 Subject: [PATCH] feat(tracker): add alpha calibration and dynamic pricing Add TOPLOC honest-noise calibration storage/dispatch and validator divergence reporting for AH-021. Add opt-in HuggingFace marketplace pricing refresh, price-change history, CLI flags, and AH-023 tracking docs. Verification: .venv/bin/python -m pytest tests/ -q -k 'not integration' => 346 passed, 2 skipped, 1 deselected; compileall packages tests passed; focused AH-021/AH-023 tests 32 passed. --- .claude/memory/alpha-hardening-navigation.md | 12 +- .scratch/alpha-hardening/README.md | 7 +- .../21-honest-noise-calibration-corpus.md | 35 +- .../issues/23-dynamic-hf-pricing.md | 53 +++ .scratch/alpha-hardening/prd.json | 25 +- .../tracker/meshnet_tracker/calibration.py | 223 ++++++++++ packages/tracker/meshnet_tracker/cli.py | 59 +++ .../tracker/meshnet_tracker/hf_pricing.py | 314 ++++++++++++++ .../meshnet_tracker/model_presets.json | 2 + packages/tracker/meshnet_tracker/server.py | 389 +++++++++++++++++- packages/validator/README.md | 25 +- .../validator/meshnet_validator/__init__.py | 8 +- packages/validator/meshnet_validator/audit.py | 82 +++- tests/test_hf_pricing.py | 150 +++++++ tests/test_hf_pricing_dispatch.py | 142 +++++++ tests/test_toploc_audit.py | 74 +++- tests/test_toploc_calibration.py | 80 ++++ tests/test_toploc_calibration_dispatch.py | 341 +++++++++++++++ 18 files changed, 1996 insertions(+), 25 deletions(-) create mode 100644 .scratch/alpha-hardening/issues/23-dynamic-hf-pricing.md create mode 100644 packages/tracker/meshnet_tracker/calibration.py create mode 100644 packages/tracker/meshnet_tracker/hf_pricing.py create mode 100644 tests/test_hf_pricing.py create mode 100644 tests/test_hf_pricing_dispatch.py create mode 100644 tests/test_toploc_calibration.py create mode 100644 tests/test_toploc_calibration_dispatch.py diff --git a/.claude/memory/alpha-hardening-navigation.md b/.claude/memory/alpha-hardening-navigation.md index b5758d5..057c33c 100644 --- a/.claude/memory/alpha-hardening-navigation.md +++ b/.claude/memory/alpha-hardening-navigation.md @@ -16,9 +16,17 @@ Active workstream (started 2026-07-04): alpha hardening of the money/trust path. **Auth foundation now available (commit 81719ed):** `packages/tracker/meshnet_tracker/auth.py` = hive HMAC (`sign_hive_request`/`verify_hive_request`, X-Meshnet-Hive-Signature/Timestamp, 300s skew) + `is_validator_token`. In the handler: `_require_role("admin"|"validator")`, `_resolve_identity()` (validator token / admin session / client-key→no-role), `_read_hive_authenticated_body()`. `TrackerServer(validator_service_token=, hive_secret=)` also read from MESHNET_VALIDATOR_SERVICE_TOKEN / MESHNET_HIVE_SECRET; CLI `--validator-service-token` / `--hive-secret`. Outgoing gossip signed via `_push_to_peers`. Tests use these fixtures — reuse the pattern in 05/03/04. -**Remaining work classification:** 12–15 are multi-tracker money/Raft ordering hardening deferred beyond single-settlement alpha; 17 needs human approval for canonical duplicate US-020 renumbering; 21 is the honest-noise calibration corpus and gates production audit thresholds, not an alpha code release. For current verification, full pytest is blocked only by local port 7000 already occupied by `meshnet-node`; excluding that environmental test passes 316 and skips 3. +**Remaining work classification:** 12–15 are multi-tracker money/Raft ordering hardening deferred beyond single-settlement alpha; 17 needs human approval for canonical duplicate US-020 renumbering. Full pytest suite re-verified 2026-07-06: 317 passed, 3 skipped, clean. -**Ralph note:** `scripts/ralph_progress.py` tracks `docs/prd.json` (35/35 done) and does NOT see `.scratch/alpha-hardening/issues/`. No ralph loop is running and no `.ralph-tui/` state exists. To use the ralph dashboard for the alpha phase, migrate the ready-for-agent alpha issues into prd.json first; otherwise track via the `Status:` header in each issue file. Do NOT use `ralph auto --parallel` on server.py-touching issues. +**Launch-readiness grilling (2026-07-06):** Locked launch plan — devnet dev/test run now, then **real mainnet SOL/USDT** (not devnet, not a new public token) for the first cohort: friends (API clients) + hired VPS/VPC hosts (our own test infra, not third-party volunteers — stake-free, risk-free if something breaks, not a long-term topology). Pricing: clients are the only party spending real money; nodes only accumulate off-chain credit and get paid in batches (30min dev / 24h later) — a failed distribution leaves funds parked, not lost, so mainnet-vs-devnet mixups are lower-risk than initially assumed. TAI token: do NOT issue/list now — ADR-0002 already locks listing behind $50k volume + 25 nodes/15 wallets plus an unresolved securities-review gate; only a dormant mainnet mint (cheap, ~few $ SOL) for name/branding reservation is in scope, bundled with treasury-key work, not before it. Treasury custody: bare keypair file (current runbook 02) is not acceptable for real funds — plan is **free native SPL multisig** (`spl-token create-multisig`, no protocol fee unlike Squads' 0.5 SOL), 2-of-3 signers, at least one cold/offline, others one-per-hired-VPS-provider to avoid correlated compromise (not yet built — ops task, no issue filed). Stake/slash asymmetry (registry/slash is a local Python adapter per ADR-0007, not on-chain) accepted for now since hired hosts are our own infra and friends aren't node operators — revisit before opening to real third-party node operators. A mainnet-vs-devnet boot guardrail was proposed and explicitly declined by the owner given the safe-by-default money flow above. + +**Two new issues from this session, both `ready-for-agent`:** +- **21 — Honest-noise calibration corpus** (`.scratch/alpha-hardening/issues/21-honest-noise-calibration-corpus.md`) rescoped from "prod gate" to a **hard alpha-release blocker**. Confirmed by code read: `verify_activation_proofs()` (`packages/validator/meshnet_validator/audit.py:94-127`) returns bool only, no raw divergence value; fleet-dispatch exists but wrong shape (`server.py:2998-3104`, pinned routes + latency, not full-fleet + TOPLOC divergence); storage wrong shape (`registry_events` has no divergence/hardware columns). Three-part build: (1) surface raw TOPLOC distance from audit.py, (2) extend dispatch to hit every registered node with fixed prompt/seed, (3) new SQLite table keyed by node+GPU+dtype. Small-fleet exception granted (N = actual hired-VPS fleet size). Hired VPS hosts stay stake-free until this closes. +- **23 — Dynamic HF-benchmarked pricing** (`.scratch/alpha-hardening/issues/23-dynamic-hf-pricing.md`), high priority but not a release blocker. Pricing today is 100% static (`DEFAULT_PRICE_PER_1K_TOKENS = 0.02`, `billing.py:21`; `model_presets.json` has no per-model price). Target: 80% of cheapest comparable provider on `https://huggingface.co/inference/models` (per-provider-per-model marketplace, `?search=` query param works, no confirmed JSON API — plain scrape attempted first, escalate to headless browser only if the table isn't in raw HTML). Human-verified `hf_aliases` + `hf_verified_match_note` (params/quantization) per model, not auto-discovered matching. Reuses the `_settlement_loop` daemon-thread pattern for a daily refresh; falls back silently to the static default on any failure. + +Both are already migrated into `.scratch/alpha-hardening/prd.json` (AH-021 updated, AH-023 added) and the README index — ready for Ralph to pick up unattended. + +**Ralph note:** `scripts/ralph_progress.py` tracks `docs/prd.json` (35/35 done) and does NOT see `.scratch/alpha-hardening/issues/`. No ralph loop is running and no `.ralph-tui/` state exists. `.scratch/alpha-hardening/prd.json` now has 23 stories (AH-001…AH-023); point Ralph at that file for the alpha-hardening branch. Do NOT use `ralph auto --parallel` on server.py-touching issues — 21 and 23 both touch `server.py`/`billing.py`/`audit.py`; if run in the same Ralph pass, run them serially, not in parallel (merge-conflict risk, same lesson as 03/04 previously). **Why:** three audits agreed the alpha blockers are unauthenticated gossip (anyone can inject billing events), the free-credit faucet, and ephemeral bans. **How to apply:** work test-first per issue acceptance criteria; use `.venv`; `cryptography` belongs in node deps (wallet.py imports it — causes many of the 24 "failures" in a fresh env). See [[project-status]] and [[autonomous-work-style]]. diff --git a/.scratch/alpha-hardening/README.md b/.scratch/alpha-hardening/README.md index 4758d6c..65df940 100644 --- a/.scratch/alpha-hardening/README.md +++ b/.scratch/alpha-hardening/README.md @@ -1,6 +1,11 @@ # Alpha hardening — planning index -Pre-release alpha audit + grilling (2026-07-04). **Research complete; planning complete; Bucket 1 blockers next.** +Pre-release alpha audit + grilling (2026-07-04). Bucket 1 trust-boundary blockers + fraud arc: **done** (16/22 original issues). Bucket 2 (12-15, multi-tracker) and 17 (doc dedup) remain deferred/human-gated — not launch blockers. + +**Launch-readiness grilling (2026-07-06):** locked plan is devnet dev/test run now, then real mainnet SOL/USDT for the first cohort — friends (API clients) + hired VPS/VPC hosts (own test infra, not third-party volunteers, stake-free). No new public token; TAI stays dormant per ADR-0002's existing volume/legal gates. Two new issues came out of this session: + +- **[21 — Honest-noise calibration corpus](./issues/21-honest-noise-calibration-corpus.md)** — rescoped from "prod gate" to a hard **alpha-release blocker**. `Status: ready-for-human` — engineering (audit.py raw divergence, tracker dispatch endpoint, SQLite corpus, p99 envelope) done 2026-07-06; blocked on a human running the calibration job against the real hired-VPS fleet before launch. +- **[23 — Dynamic HF-benchmarked pricing](./issues/23-dynamic-hf-pricing.md)** — new, high priority but not a release blocker. `Status: done` — engineering complete 2026-07-06 (hf_pricing.py, opt-in daily refresh loop, GET /v1/pricing/hf/history); real `hf_aliases` curation per model is a follow-up human sign-off, not a completion blocker. Locked scope: one settlement tracker, open node join, devnet mock-USDT, reputation carries forward → fraud must be bounded. See [ADR-0016](../../docs/adr/0016-alpha-scope-and-known-limitations.md). diff --git a/.scratch/alpha-hardening/issues/21-honest-noise-calibration-corpus.md b/.scratch/alpha-hardening/issues/21-honest-noise-calibration-corpus.md index a3d1f47..9a080a9 100644 --- a/.scratch/alpha-hardening/issues/21-honest-noise-calibration-corpus.md +++ b/.scratch/alpha-hardening/issues/21-honest-noise-calibration-corpus.md @@ -1,28 +1,42 @@ Status: ready-for-human +**BLOCKS ALPHA RELEASE.** Scoped 2026-07-06 during alpha-launch-readiness grilling session — must complete before real-money (mainnet SOL/USDT) traffic goes live for the friends + hired-VPS-host launch. Loose/uncalibrated thresholds + manual admin slash-reversal are the stopgap only until this closes. + +**Engineering complete 2026-07-06; blocked on a human running it against the real hired-VPS fleet before launch.** The three code gaps below are closed and unit-tested (see Deliverables), but nothing in a dev session can stand in for actually dispatching the job at real hardware — that step, plus the threshold/FPR write-up that depends on its output, needs an operator with the live fleet. See the validator README's "Honest-noise calibration corpus" section for the operational how-to. + # 21 — Honest-noise TOPLOC calibration corpus ## What to build -Before enabling production TOPLOC audit thresholds, collect an **honest-noise baseline** across the heterogeneous volunteer fleet. Run identical inference jobs on every active node/GPU combo; measure the divergence envelope (TOPLOC exponent/mantissa deltas, logprob-rank spread) under real hardware variance. +Before enabling production TOPLOC audit thresholds, collect an **honest-noise baseline** across the active fleet. Run identical inference jobs on every active node/GPU combo; measure the divergence envelope (TOPLOC exponent/mantissa deltas, logprob-rank spread) under real hardware variance. This must be driven by the tracker (scheduled/dispatched job), not a manual one-off script, so it can be re-run as the fleet's hardware mix changes. Per [ADR-0018 consequences](../../docs/adr/0018-fraud-detection-verification-and-reputation.md): threshold calibration requires an honest-noise corpus across the fleet before production thresholds. Research anchor: `.scratch/alpha-hardening/research-verifiable-inference.md` §8 layer 3 — "collect this first — run identical jobs across the current node fleet to measure the honest divergence envelope before setting thresholds." +**Launch context (why this is buildable now, not a research project):** first-launch nodes are hired VPS/VPC hosts under our own direct control (test infrastructure we pay for, not third-party volunteers) — not a long-term topology, but risk-free for calibration purposes since there's no external party to dispute a bad reading. Friends are client-side users of the API in this phase, not node operators. Run the calibration pass against this small, fully-controlled fleet first; hired hosts stay stake-free until it's done, then move to real staking once thresholds derive from their own hardware. + +**Current gap (confirmed 2026-07-06 by code read):** none of the three pieces below exist yet. + +1. `verify_activation_proofs()` (`packages/validator/meshnet_validator/audit.py:94-127`) returns a **plain bool** — no raw TOPLOC divergence/distance value is ever computed or surfaced. Every "done" fraud-detection issue (06–10) currently runs on a guessed threshold baked into that bool, not a calibrated one. +2. Fleet dispatch exists but is the wrong shape: `_handle_benchmark_hop_penalty` / `_handle_benchmark_results` (`packages/tracker/meshnet_tracker/server.py:2998-3104`, from the old US-030 latency work) targets pinned 1–3-node *routes* and measures latency, not TOPLOC divergence across *every* registered node. +3. Storage is the wrong shape: `record_audit_outcome` (`packages/contracts/meshnet_contracts/__init__.py:416`) persists only `strike_count`/`banned`/`passed` to `registry_events` — no divergence value, no GPU/dtype/hardware-profile column anywhere. Benchmark results otherwise land in a flat JSON file (`server.benchmark_results_path`), not a queryable per-node/hardware schema. + ## Deliverables -- [ ] Scripted benchmark job (fixed prompt, model preset, seed policy) runnable on all nodes -- [ ] Aggregated corpus artifact (per node: GPU model, dtype, TOPLOC deltas vs reference) -- [ ] Recommended tolerance thresholds documented (p99 honest envelope + safety margin) -- [ ] Gate checklist: production audit enable blocked until corpus covers ≥N distinct hardware profiles (define N in runbook, suggest ≥3) +- [x] Extend the TOPLOC verify call path (`audit.py`) to return the raw distance/divergence metric alongside the existing bool — `verify_activation_proofs_detailed()` / `ToplocVerificationResult` in `packages/validator/meshnet_validator/audit.py`; `verify_activation_proofs()` kept as a thin bool-only wrapper for existing callers. Also fixes a real bug this issue's code-read surfaced: the old code did `bool(_call_toploc(...))`, which is always `True` for the real `toploc` library's non-empty per-chunk `VerificationResult` list regardless of divergence — `tests/test_toploc_audit.py::test_verify_activation_proofs_detailed_aggregates_per_chunk_divergence` exercises this directly. +- [x] Extend the existing fleet-dispatch pattern (`server.py:2998+`) from pinned-route benchmarking to a tracker-scheduled job that hits **every currently registered node** with a fixed prompt/model/seed — `POST /v1/calibration/toploc/run` (admin/validator-gated, same shape as `POST /v1/benchmark/hop-penalty`) in `packages/tracker/meshnet_tracker/server.py`. Dispatches to every node that can solo-serve the full model range (single-hop pinned route, isolating one node's hardware noise from route-composition effects); partial-shard nodes are reported under `skipped_partial_shard_node_ids`, and nodes that don't answer the on-demand TOPLOC commitment fetch are reported per-node under `"skipped": "..."` rather than counted as pass or fail. See `tests/test_toploc_calibration_dispatch.py`. +- [x] Add a small SQLite table (same pattern as `billing.py`/`accounts.py`) keyed by node wallet + GPU model + dtype, storing the divergence value per calibration run — `packages/tracker/meshnet_tracker/calibration.py::ToplocCalibrationStore`, `toploc_calibration_runs` table. +- [x] Aggregation: p99 honest envelope + safety margin computed from that table, written as the recommended tolerance constants — `ToplocCalibrationStore.envelope()`, exposed via `GET /v1/calibration/toploc/results`. +- [x] Gate checklist: production audit enable blocked until corpus covers ≥N distinct hardware profiles — `ToplocCalibrationStore.gate_status(min_hardware_profiles=N)`; N is `--toploc-calibration-gate-min-hardware-profiles` (default 1) on the tracker CLI, documented alpha exception in the validator README. ## Acceptance criteria -- [ ] Corpus collected from current fleet (or documented subset + extrapolation note) -- [ ] Threshold constants in validator config derived from corpus, not guessed -- [ ] False-positive rate estimate documented at chosen thresholds -- [ ] README / runbook cross-link: **do not enable production audits** until this issue closes +- [ ] Corpus collected from the current hired-VPS fleet (documented as a small-fleet alpha corpus, not the eventual volunteer-fleet corpus) — **not done: needs a human to run `POST /v1/calibration/toploc/run` against the live hired-VPS fleet before launch; no such fleet exists in a dev session.** +- [ ] Threshold constants in validator config derived from corpus, not guessed — mechanically ready (`envelope()` returns them) but depends on the real corpus above; not yet wired into `ToplocAuditConfig` as enforced thresholds (deliberately — enforcing unvalidated thresholds would be worse than today's guessed bool). +- [ ] False-positive rate estimate documented at chosen thresholds — `envelope()` returns `estimated_false_positive_rate` (in-sample: fraction of the recorded corpus the recommended thresholds would themselves flag); needs the real corpus to be a meaningful number, and should be written up in the runbook once collected. +- [x] README / runbook cross-link: **do not enable production audits** until this issue closes — `packages/validator/README.md` "TOPLOC audit contract" section, updated with the full operational how-to. +- [x] Note in the runbook that this alpha corpus must be re-run once the fleet grows beyond the hired-VPS set (different hardware mix invalidates the envelope) — same README section. ## ADR links @@ -30,8 +44,9 @@ Research anchor: `.scratch/alpha-hardening/research-verifiable-inference.md` §8 ## Blocked by -- `06-fraud-toploc-integration.md` (TOPLOC wired; calibration uses same primitive) +- `06-fraud-toploc-integration.md` (TOPLOC wired; calibration uses same primitive) — done ## Blocks (prod gate) +- Alpha release to real-money friends+hired-VPS launch (raised from "production adaptive audit thresholds" to a hard alpha-release gate during 2026-07-06 grilling) - Production enable of adaptive audit thresholds (issues 09–10 in prod) diff --git a/.scratch/alpha-hardening/issues/23-dynamic-hf-pricing.md b/.scratch/alpha-hardening/issues/23-dynamic-hf-pricing.md new file mode 100644 index 0000000..bd79520 --- /dev/null +++ b/.scratch/alpha-hardening/issues/23-dynamic-hf-pricing.md @@ -0,0 +1,53 @@ +Status: done + +Scoped 2026-07-06 during alpha-launch-readiness grilling session. High priority, ship-soon for launch — **not** an alpha-release blocker (unlike issue 21): a stale/static price is a revenue/business-model risk, not a safety risk, so the friends + hired-VPS launch may proceed on the current static default while this lands in parallel. + +# 23 — Dynamic per-model pricing benchmarked against HuggingFace inference rates + +## What to build + +Client-facing price per model should track the market: **80% of the cheapest comparable provider rate on HuggingFace's inference marketplace** (`https://huggingface.co/inference/models`), refreshed daily, auto-adjusting so served models stay competitively priced as the market moves. Nodes are unaffected by this loop (per launch design: clients are the only party spending real money; node payouts come from the 90/10 split of whatever price is charged, per ADR-0015/`packages/validator/README.md`). + +**Current state (confirmed by code read 2026-07-06):** pricing is 100% static today. `DEFAULT_PRICE_PER_1K_TOKENS = 0.02` (`packages/tracker/meshnet_tracker/billing.py:21`) is the fallback nearly every model hits, since `model_presets.json` currently has no `price_per_1k_tokens` key for any preset. `BillingLedger.set_price(model, price)` (`billing.py:67-69`) is the only write path and already exists — no CLI/admin route calls it yet. No external HTTP/market-data integration exists anywhere in the tracker. + +**Data source:** `https://huggingface.co/inference/models` aggregates multiple providers (novita, together, fireworks-ai, deepinfra, etc.) with per-model, per-provider $/1M input and output token pricing; the "cheapest" badge already identifies the lowest-cost provider per model on the page itself. It supports a GET query param for filtering, e.g. `?search=GLM`. **No confirmed public JSON API was found** during this session's fetch — the page reads as a rendered table. Owner's suggestion: try a plain `requests` + BeautifulSoup scrape first; if the pricing table turns out to be client-rendered (not present in the initial HTML), that's the fallback signal to escalate to a headless-browser fetch (e.g. Playwright) — confirm which is needed during implementation before building the full pipeline around it. Another data source is acceptable if more convenient/stable, owner is not wedded to this specific page. + +## Deliverables + +- [x] Live-fetch attempt (requests + BeautifulSoup against the HF page with `?search=`, or an equivalent stable source) as the primary path — confirm during implementation whether the pricing table is present in the raw HTML or requires a headless-browser fetch, and note which in the PR +- [x] Extend `model_presets.json` per model with: `hf_aliases` (curated list of comparable HF model+provider IDs — **human-verified, not auto-discovered**), `hf_verified_match_note` (free text: params count + quantization confirmation, so a human signs off once per alias that it is a fair comparable before it's used for auto-pricing), `hf_last_price_per_1k` (derived from the $/1M rate), `hf_last_updated` (ISO date) +- [x] Daily refresh job reusing the tracker's existing daemon-thread pattern (`_settlement_loop`/`_deposit_loop` in `server.py`, `threading.Event().wait(interval)` loop) — for each preset with a non-empty `hf_aliases` list, fetch current pricing for those aliases, compute `0.8 × cheapest matched alias price`, call `set_price()`, and update `hf_last_price_per_1k`/`hf_last_updated` +- [x] Every price change logged (old price, new price, source alias, timestamp) — needed for dispute auditability if a client questions a charge +- [x] Fallback behavior: empty/missing `hf_aliases`, fetch failure, or no verified match → silently keep the existing static default price. Never error the pricing path, never zero-price a model + +## Acceptance criteria + +- [x] At least one model preset has a working end-to-end refresh (alias → live fetch → 80% computed price → `set_price()` called → metadata updated) demonstrated in a test +- [x] Models without a curated/verified alias continue to use the static default, unaffected by this feature +- [x] Fetch failures (network error, page structure change, no match found) degrade gracefully — logged, not raised to the request path +- [x] Price-change log is queryable/inspectable (doesn't need a UI yet — a log line or table row is sufficient for alpha) +- [x] Note in the runbook/issue on which fetch mechanism (plain HTTP scrape vs. headless browser) was actually required, so the next person doesn't have to rediscover it + +## Implementation notes (2026-07-06) + +**Fetch mechanism confirmed: plain HTTP scrape, no headless browser needed.** Live-fetched `https://huggingface.co/inference/models?search=GLM` this session — the pricing table is server-rendered into the initial HTML response (SvelteKit SSR), confirmed by grepping the raw response for `cheapest`/`$`-prefixed price cells before any JS runs. A stdlib `urllib.request` GET + `html.parser.HTMLParser`-based table walk is sufficient; no `requests`/`bs4`/Playwright dependency was added, matching this package's existing zero-new-HTTP-dependency convention (`gossip.py`/`raft.py`/`server.py` all use `urllib.request` only). Each row's most stable extraction anchor turned out to be the `` link, not the display text (which duplicates the repo id at two responsive breakpoints and is easy to mis-parse). + +**What shipped:** new `packages/tracker/meshnet_tracker/hf_pricing.py` — pure HTML parser (`parse_hf_pricing_table`), alias matching (`cheapest_matching_quote`, supports both `org/repo` and `org/repo::provider` forms so a human can pin a specific provider's deployment), a pure per-preset computation function (`refresh_preset_price`, never raises), and `HfPricingLog` (SQLite-backed change log, same shape as `billing.py`/`calibration.py`). `TrackerServer` gained an opt-in (`enable_hf_pricing=True` / `--enable-hf-pricing`) daily daemon thread (`_hf_pricing_loop`, same `threading.Event().wait(interval)` shape as `_settlement_loop`) and `GET /v1/pricing/hf/history` (admin/validator-gated, mirrors `/v1/calibration/toploc/results`). `model_presets.json`'s `kimi-k2.7` preset now carries the `hf_aliases`/`hf_verified_match_note` schema fields, left as an empty list pending a human sign-off on a genuinely comparable HF listing (params count + quantization) — per this issue's own "human-verified, not auto-discovered" requirement, an agent should not fabricate that sign-off. This also means the shipped default config demonstrates the required "no alias → static price, unaffected" fallback for a real production preset; the alias→live-fetch→80%→set_price() path is demonstrated end-to-end against an injected fetch backend in `tests/test_hf_pricing_dispatch.py` (the `fetch_html=`/`hf_pricing_fetch_html=` injection point mirrors this codebase's `backend=` convention for anything that would otherwise hit the network in tests). + +**Bug caught and fixed while wiring this in:** `TrackerServer` previously did `dict(DEFAULT_MODEL_PRESETS)` when no explicit `model_presets` was passed — a shallow copy that aliases every preset's inner dict to the shared module-level global. Writing `hf_last_price_per_1k`/`hf_last_updated` in place would have leaked across every other `TrackerServer` instance in the same process (real risk in the test suite, and in any future multi-tracker-in-one-process embedding). Fixed with a `_clone_model_presets()` helper that also shallow-copies each preset dict. + +**Follow-up for a human (not a completion blocker):** populate real `hf_aliases`/`hf_verified_match_note` entries for production presets once someone has confirmed a genuinely comparable HF-listed deployment (params + quantization) — that activates dynamic pricing for that model on the next refresh tick. Until then every preset safely stays on its static price. + +Tests: `tests/test_hf_pricing.py` (11 tests: parsing, blended-price math, alias matching incl. provider-scoped aliases, all three fallback paths, log persistence) + `tests/test_hf_pricing_dispatch.py` (5 tests: full TrackerServer end-to-end refresh, unaffected-without-alias, history auth gating, history content, history model filter). Full suite (`pytest tests/ -q -k "not integration"`): 346 passed, 2 skipped. + +## ADR links + +- [ADR-0015](../../docs/adr/0015-usdt-custodial-settlement.md) — settlement/pricing this touches (90/10 split, per-model pricing) + +## Blocked by + +None — independent of the alpha-hardening trust-boundary work; touches `billing.py`/`server.py` pricing paths only. + +## Blocks + +None — ship-soon for launch quality, not a release gate (see status note above). diff --git a/.scratch/alpha-hardening/prd.json b/.scratch/alpha-hardening/prd.json index e704fbb..f53aedf 100644 --- a/.scratch/alpha-hardening/prd.json +++ b/.scratch/alpha-hardening/prd.json @@ -441,11 +441,11 @@ ], "priority": 21, "passes": true, - "notes": "Source issue: .scratch/alpha-hardening/issues/21-honest-noise-calibration-corpus.md\nRalph skip: Source issue is ready-for-human/deferred; skipped by unattended Ralph auto.", + "notes": "Source issue: .scratch/alpha-hardening/issues/21-honest-noise-calibration-corpus.md. BLOCKS ALPHA RELEASE (real-money friends+hired-VPS launch) — rescoped 2026-07-06, no longer a Ralph-skip.", "dependsOn": [ "AH-006" ], - "completionNotes": "Source issue is ready-for-human/deferred; skipped by unattended Ralph auto." + "completionNotes": "Engineering complete and unit-tested (validator audit.py detailed-verify aggregation, tracker calibration.py corpus store, calibration dispatch endpoints). Marked ready-for-human, not done: real corpus collection against the live hired-VPS fleet, and the threshold/FPR write-up that depends on its output, need a human operator — see .ralph-tui/progress.md and packages/validator/README.md." }, { "id": "AH-022", @@ -464,9 +464,28 @@ "notes": "Source issue: .scratch/alpha-hardening/issues/22-doc-memory-project-status.md", "dependsOn": [], "completionNotes": "Marked done in source alpha-hardening issue file before PRD generation." + }, + { + "id": "AH-023", + "title": "23 — Dynamic per-model pricing benchmarked against HuggingFace inference rates", + "description": "Status: ready-for-agent\n\nScoped 2026-07-06 during alpha-launch-readiness grilling session. High priority, ship-soon for launch — **not** an alpha-release blocker (unlike issue 21): a stale/static price is a revenue/business-model risk, not a safety risk, so the friends + hired-VPS launch may proceed on the current static default while this lands in parallel.\n\n# 23 — Dynamic per-model pricing benchmarked against HuggingFace inference rates\n\n## What to build\n\nClient-facing price per model should track the market: **80% of the cheapest comparable provider rate on HuggingFace's inference marketplace** (`https://huggingface.co/inference/models`), refreshed daily, auto-adjusting so served models stay competitively priced as the market moves. Nodes are unaffected by this loop (per launch design: clients are the only party spending real money; node payouts come from the 90/10 split of whatever price is charged, per ADR-0015/`packages/validator/README.md`).\n\n**Current state (confirmed by code read 2026-07-06):** pricing is 100% static today. `DEFAULT_PRICE_PER_1K_TOKENS = 0.02` (`packages/tracker/meshnet_tracker/billing.py:21`) is the fallback nearly every model hits, since `model_presets.json` currently has no `price_per_1k_tokens` key for any preset. `BillingLedger.set_price(model, price)` (`billing.py:67-69`) is the only write path and already exists — no CLI/admin route calls it yet. No external HTTP/market-data integration exists anywhere in the tracker.\n\n**Data source:** `https://huggingface.co/inference/models` aggregates multiple providers (novita, together, fireworks-ai, deepinfra, etc.) with per-model, per-provider $/1M input and output token pricing; the \"cheapest\" badge already identifies the lowest-cost provider per model on the page itself. It supports a GET query param for filtering, e.g. `?search=GLM`. **No confirmed public JSON API was found** during this session's fetch — the page reads as a rendered table. Owner's suggestion: try a plain `requests` + BeautifulSoup scrape first; if the pricing table turns out to be client-rendered (not present in the initial HTML), that's the fallback signal to escalate to a headless-browser fetch (e.g. Playwright) — confirm which is needed during implementation before building the full pipeline around it. Another data source is acceptable if more convenient/stable, owner is not wedded to this specific page.\n\n## Deliverables\n\n- [x] Live-fetch attempt (requests + BeautifulSoup against the HF page with `?search=`, or an equivalent stable source) as the primary path — confirm during implementation whether the pricing table is present in the raw HTML or requires a headless-browser fetch, and note which in the PR\n- [x] Extend `model_presets.json` per model with: `hf_aliases` (curated list of comparable HF model+provider IDs — **human-verified, not auto-discovered**), `hf_verified_match_note` (free text: params count + quantization confirmation, so a human signs off once per alias that it is a fair comparable before it's used for auto-pricing), `hf_last_price_per_1k` (derived from the $/1M rate), `hf_last_updated` (ISO date)\n- [x] Daily refresh job reusing the tracker's existing daemon-thread pattern (`_settlement_loop`/`_deposit_loop` in `server.py`, `threading.Event().wait(interval)` loop) — for each preset with a non-empty `hf_aliases` list, fetch current pricing for those aliases, compute `0.8 × cheapest matched alias price`, call `set_price()`, and update `hf_last_price_per_1k`/`hf_last_updated`\n- [x] Every price change logged (old price, new price, source alias, timestamp) — needed for dispute auditability if a client questions a charge\n- [x] Fallback behavior: empty/missing `hf_aliases`, fetch failure, or no verified match → silently keep the existing static default price. Never error the pricing path, never zero-price a model\n\n## Acceptance criteria\n\n- [x] At least one model preset has a working end-to-end refresh (alias → live fetch → 80% computed price → `set_price()` called → metadata updated) demonstrated in a test\n- [x] Models without a curated/verified alias continue to use the static default, unaffected by this feature\n- [x] Fetch failures (network error, page structure change, no match found) degrade gracefully — logged, not raised to the request path\n- [x] Price-change log is queryable/inspectable (doesn't need a UI yet — a log line or table row is sufficient for alpha)\n- [x] Note in the runbook/issue on which fetch mechanism (plain HTTP scrape vs. headless browser) was actually required, so the next person doesn't have to rediscover it\n\n## ADR links\n\n- [ADR-0015](../../docs/adr/0015-usdt-custodial-settlement.md) — settlement/pricing this touches (90/10 split, per-model pricing)\n\n## Blocked by\n\nNone — independent of the alpha-hardening trust-boundary work; touches `billing.py`/`server.py` pricing paths only.\n\n## Blocks\n\nNone — ship-soon for launch quality, not a release gate (see status note above).\n", + "acceptanceCriteria": [ + "End-to-end refresh demonstrated for at least one model preset (alias -> live fetch -> 80% computed price -> set_price() -> metadata updated)", + "Models without curated/verified alias continue using static default, unaffected", + "Fetch failures degrade gracefully (logged, not raised to request path)", + "Price-change log is queryable/inspectable (log line or table row sufficient for alpha)", + "Note in issue which fetch mechanism (plain scrape vs headless browser) was actually required", + "Update the source issue file Status header to done when complete", + "Run relevant pytest tests; run the full suite when practical or document why not" + ], + "priority": 23, + "passes": true, + "notes": "Source issue: .scratch/alpha-hardening/issues/23-dynamic-hf-pricing.md. High priority, ship-soon for launch, NOT an alpha-release blocker (unlike AH-021).", + "dependsOn": [], + "completionNotes": "Completed by agent" } ], "metadata": { - "updatedAt": "2026-07-05T16:21:45.488Z" + "updatedAt": "2026-07-06T06:01:25.474Z" } } \ No newline at end of file diff --git a/packages/tracker/meshnet_tracker/calibration.py b/packages/tracker/meshnet_tracker/calibration.py new file mode 100644 index 0000000..e43e560 --- /dev/null +++ b/packages/tracker/meshnet_tracker/calibration.py @@ -0,0 +1,223 @@ +"""TOPLOC honest-noise calibration corpus (ADR-0018 consequences, issue 21). + +Production TOPLOC audit thresholds must be derived from an empirical +honest-noise baseline across the active fleet's hardware, not guessed +(research-verifiable-inference.md §8 layer 3). This store persists one row +per (node wallet, GPU model, dtype) calibration run so thresholds can be +computed from a queryable corpus instead of a flat JSON dump, and re-derived +whenever the fleet's hardware mix changes. +""" + +from __future__ import annotations + +import json +import sqlite3 +import threading +import time +import uuid + +DEFAULT_CALIBRATION_DB_PATH = "toploc_calibration.sqlite" +# Headroom added on top of the observed p99 honest-noise envelope so normal +# hardware variance doesn't trip the recommended threshold (ADR-0018 §3). +DEFAULT_SAFETY_MARGIN = 0.20 +DEFAULT_PERCENTILE = 0.99 + + +class ToplocCalibrationStore: + """Thread-safe SQLite-backed corpus of per-node TOPLOC divergence runs.""" + + def __init__(self, db_path: str | None = None) -> None: + self._db_path = db_path + self._lock = threading.Lock() + self._runs: list[dict] = [] + if self._db_path: + self._init_db() + self._load_from_db() + + def record_run( + self, + *, + node_wallet: str, + gpu_model: str | None, + dtype: str | None, + model: str, + passed: bool, + exp_intersections: float | None, + mant_err_mean: float | None, + mant_err_median: float | None, + ts: float | None = None, + ) -> dict: + run = { + "id": f"cal-{uuid.uuid4().hex}", + "node_wallet": node_wallet, + "gpu_model": gpu_model or "unknown", + "dtype": dtype or "unknown", + "model": model, + "passed": bool(passed), + "exp_intersections": exp_intersections, + "mant_err_mean": mant_err_mean, + "mant_err_median": mant_err_median, + "ts": ts if ts is not None else time.time(), + } + with self._lock: + self._runs.append(run) + self._save_run(run) + return run + + def runs(self) -> list[dict]: + with self._lock: + return list(self._runs) + + def distinct_hardware_profiles(self) -> set[tuple[str, str]]: + with self._lock: + return {(r["gpu_model"], r["dtype"]) for r in self._runs} + + def gate_status(self, *, min_hardware_profiles: int) -> dict: + """Whether the corpus is broad enough to enable production thresholds. + + Alpha exception (issue 21): with a small, fully hired/controlled VPS + fleet, ``min_hardware_profiles`` may legitimately equal the fleet's + actual distinct hardware count — this must be revisited before a + public/volunteer launch broadens the hardware mix. + """ + distinct = len(self.distinct_hardware_profiles()) + return { + "distinct_hardware_profiles": distinct, + "min_hardware_profiles": min_hardware_profiles, + "sample_count": len(self._runs), + "ready": distinct > 0 and distinct >= min_hardware_profiles, + } + + def envelope( + self, + *, + percentile: float = DEFAULT_PERCENTILE, + safety_margin: float = DEFAULT_SAFETY_MARGIN, + ) -> dict: + """Recommended tolerance constants derived from the corpus. + + `exp_intersections` (higher = better match) gets a floor at its + worst-case (low) percentile minus margin; the mantissa errors + (higher = worse) get a ceiling at their worst-case (high) percentile + plus margin. Returns None for a metric with no samples yet. + """ + with self._lock: + runs = list(self._runs) + exp_vals = sorted(v for r in runs if (v := r["exp_intersections"]) is not None) + mean_vals = sorted(v for r in runs if (v := r["mant_err_mean"]) is not None) + median_vals = sorted(v for r in runs if (v := r["mant_err_median"]) is not None) + min_exp = _floor(exp_vals, 1.0 - percentile, safety_margin) + max_mean = _ceiling(mean_vals, percentile, safety_margin) + max_median = _ceiling(median_vals, percentile, safety_margin) + return { + "sample_count": len(runs), + "distinct_hardware_profiles": len(self.distinct_hardware_profiles()), + "percentile": percentile, + "safety_margin": safety_margin, + "recommended_min_exp_intersections": min_exp, + "recommended_max_mant_err_mean": max_mean, + "recommended_max_mant_err_median": max_median, + # In-sample estimate only: the fraction of this same honest + # corpus that the recommended thresholds would themselves flag. + # Not a substitute for independent validation data — but a + # documented starting estimate per issue 21's acceptance + # criteria, and a sanity check that the margin isn't too tight. + "estimated_false_positive_rate": _false_positive_rate( + runs, min_exp=min_exp, max_mean=max_mean, max_median=max_median, + ), + } + + # ---- persistence (billing.py pattern) ---- + + def _init_db(self) -> None: + con = sqlite3.connect(self._db_path) # type: ignore[arg-type] + con.execute( + "CREATE TABLE IF NOT EXISTS toploc_calibration_runs " + "(run_id TEXT PRIMARY KEY, node_wallet TEXT NOT NULL, " + "gpu_model TEXT NOT NULL, dtype TEXT NOT NULL, payload TEXT NOT NULL, " + "ts REAL NOT NULL)" + ) + con.commit() + con.close() + + def _load_from_db(self) -> None: + con = sqlite3.connect(self._db_path) # type: ignore[arg-type] + rows = con.execute( + "SELECT payload FROM toploc_calibration_runs ORDER BY ts, run_id" + ).fetchall() + con.close() + for (payload,) in rows: + try: + self._runs.append(json.loads(payload)) + except json.JSONDecodeError: + continue + + def _save_run(self, run: dict) -> None: + if not self._db_path: + return + con = sqlite3.connect(self._db_path) # type: ignore[arg-type] + con.execute( + "INSERT OR IGNORE INTO toploc_calibration_runs " + "(run_id, node_wallet, gpu_model, dtype, payload, ts) VALUES (?, ?, ?, ?, ?, ?)", + (run["id"], run["node_wallet"], run["gpu_model"], run["dtype"], json.dumps(run), float(run["ts"])), + ) + con.commit() + con.close() + + +def _percentile(sorted_vals: list[float], p: float) -> float: + if len(sorted_vals) == 1: + return sorted_vals[0] + k = (len(sorted_vals) - 1) * p + lo = int(k) + hi = min(lo + 1, len(sorted_vals) - 1) + if lo == hi: + return sorted_vals[lo] + return sorted_vals[lo] + (sorted_vals[hi] - sorted_vals[lo]) * (k - lo) + + +def _floor(sorted_vals: list[float], p: float, safety_margin: float) -> float | None: + if not sorted_vals: + return None + return max(0.0, _percentile(sorted_vals, p) * (1.0 - safety_margin)) + + +def _ceiling(sorted_vals: list[float], p: float, safety_margin: float) -> float | None: + if not sorted_vals: + return None + return _percentile(sorted_vals, p) * (1.0 + safety_margin) + + +def _false_positive_rate( + runs: list[dict], + *, + min_exp: float | None, + max_mean: float | None, + max_median: float | None, +) -> float | None: + """Fraction of the (honest, by construction) corpus that would be + flagged by the recommended thresholds — an in-sample false-positive + rate estimate, not out-of-sample validation.""" + if not runs: + return None + flagged = 0 + for r in runs: + exp = r["exp_intersections"] + mean = r["mant_err_mean"] + median = r["mant_err_median"] + would_flag = ( + (min_exp is not None and exp is not None and exp < min_exp) + or (max_mean is not None and mean is not None and mean > max_mean) + or (max_median is not None and median is not None and median > max_median) + ) + if would_flag: + flagged += 1 + return flagged / len(runs) + + +__all__ = [ + "DEFAULT_CALIBRATION_DB_PATH", + "DEFAULT_SAFETY_MARGIN", + "DEFAULT_PERCENTILE", + "ToplocCalibrationStore", +] diff --git a/packages/tracker/meshnet_tracker/cli.py b/packages/tracker/meshnet_tracker/cli.py index 5b08f0f..3221479 100644 --- a/packages/tracker/meshnet_tracker/cli.py +++ b/packages/tracker/meshnet_tracker/cli.py @@ -6,6 +6,7 @@ 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 DEFAULT_REGISTRY_DB_PATH = "meshnet_registry.sqlite3" @@ -143,6 +144,55 @@ def main() -> None: "(default: MESHNET_HIVE_SECRET env; required for multi-tracker replication)" ), ) + common.add_argument( + "--toploc-calibration-db", + default=None, + metavar="PATH", + help=( + "SQLite path for the AH-021 honest-noise TOPLOC calibration corpus " + "(enables POST /v1/calibration/toploc/run + GET /v1/calibration/toploc/results)" + ), + ) + common.add_argument( + "--toploc-reference-node-url", + default=None, + help="Reference node the calibration job teacher-forces claimed tokens against (see validator README)", + ) + common.add_argument( + "--toploc-calibration-gate-min-hardware-profiles", + type=int, + default=1, + help=( + "Distinct (GPU model, dtype) profiles the corpus must cover before " + "gate_status.ready is true (alpha exception: fleet size is acceptable)" + ), + ) + common.add_argument( + "--enable-hf-pricing", + action="store_true", + help=( + "Enable the daily dynamic pricing refresh (issue 23): for presets with a " + "curated hf_aliases list, sets the client price to 80%% of the cheapest " + "matching HuggingFace inference-marketplace rate. Presets without " + "hf_aliases are unaffected and keep their static price." + ), + ) + common.add_argument( + "--hf-pricing-log-db", + default=None, + metavar="PATH", + help=( + "SQLite database path for the dynamic pricing change log " + f"(default when --enable-hf-pricing is set: {DEFAULT_HF_PRICING_LOG_DB_PATH}; " + "enables GET /v1/pricing/hf/history)" + ), + ) + common.add_argument( + "--hf-pricing-refresh-interval", + type=float, + default=86400.0, + help="Seconds between dynamic pricing refresh passes (default: daily)", + ) parser = argparse.ArgumentParser( prog="meshnet-tracker", @@ -189,6 +239,15 @@ def main() -> None: payout_dust_floor=args.payout_dust_floor, validator_service_token=args.validator_service_token, hive_secret=args.hive_secret, + toploc_calibration_db=args.toploc_calibration_db, + toploc_reference_node_url=args.toploc_reference_node_url, + toploc_calibration_gate_min_hardware_profiles=args.toploc_calibration_gate_min_hardware_profiles, + enable_hf_pricing=args.enable_hf_pricing, + hf_pricing_log_db=( + args.hf_pricing_log_db + or (DEFAULT_HF_PRICING_LOG_DB_PATH if args.enable_hf_pricing else None) + ), + hf_pricing_refresh_interval=args.hf_pricing_refresh_interval, ) port = server.start() print(f"meshnet-tracker listening on http://{args.host}:{port}", flush=True) diff --git a/packages/tracker/meshnet_tracker/hf_pricing.py b/packages/tracker/meshnet_tracker/hf_pricing.py new file mode 100644 index 0000000..94fc062 --- /dev/null +++ b/packages/tracker/meshnet_tracker/hf_pricing.py @@ -0,0 +1,314 @@ +"""Dynamic per-model pricing benchmarked against HuggingFace inference rates (issue 23). + +Client-facing price per model tracks the market: 80% of the cheapest +comparable provider rate on HuggingFace's inference marketplace +(https://huggingface.co/inference/models), refreshed daily. Nodes are +unaffected — this only ever calls ``BillingLedger.set_price`` (the ledger's +existing write path), never touches node payouts (ADR-0015's 90/10 split +still applies to whatever price is charged). + +Confirmed 2026-07-06: the pricing table is server-rendered into the initial +HTML response (SvelteKit SSR) — a plain stdlib ``urllib.request`` GET plus +HTML parsing is sufficient. No headless-browser fetch is required. Each +table row carries an anchor whose href is +``///?inference_api=true&inference_provider=``, which is +a cheaper and more stable extraction anchor than the display text (which +duplicates the repo id at two responsive breakpoints). +""" + +from __future__ import annotations + +import json +import re +import sqlite3 +import threading +import time +import urllib.parse +import urllib.request +from dataclasses import dataclass +from html.parser import HTMLParser +from typing import Callable + +HF_INFERENCE_MODELS_URL = "https://huggingface.co/inference/models" +DEFAULT_HF_PRICING_LOG_DB_PATH = "hf_pricing_log.sqlite" +DEFAULT_CLIENT_PRICE_FRACTION = 0.80 # charge 80% of the cheapest comparable rate + +_ROW_HREF_RE = re.compile( + r"^/(?P[^/]+/[^/?]+)/\?inference_api=true&inference_provider=(?P[^&\"]+)" +) +_PRICE_RE = re.compile(r"^\$[\d,]*\.?\d+$") + + +@dataclass(frozen=True) +class HfPriceQuote: + """One (model, provider) row from the HF inference pricing table.""" + + repo_id: str + provider: str + input_per_1m: float + output_per_1m: float + + def blended_price_per_1k_tokens(self) -> float: + """Average of input/output $-per-1M-token rates, converted to $/1k. + + The tracker bills a single per-1k-token rate (``BillingLedger`` + doesn't distinguish prompt vs. completion tokens), so this is the + simplest fair proxy for "this provider's rate" in that unit. + """ + return (self.input_per_1m + self.output_per_1m) / 2.0 / 1000.0 + + def alias_keys(self) -> tuple[str, str]: + """Both the bare-repo and repo::provider forms an ``hf_aliases`` entry may use.""" + return (self.repo_id.lower(), f"{self.repo_id.lower()}::{self.provider.lower()}") + + +class _HfPricingTableParser(HTMLParser): + """Extracts (repo_id, provider, input$/1M, output$/1M) rows from the raw HTML.""" + + def __init__(self) -> None: + super().__init__() + self._in_tr = False + self._row_match: tuple[str, str] | None = None + self._row_prices: list[float] = [] + self._in_td = False + self._td_text: list[str] = [] + self.quotes: list[HfPriceQuote] = [] + + def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None: + if tag == "tr": + self._in_tr = True + self._row_match = None + self._row_prices = [] + elif tag == "a" and self._in_tr and self._row_match is None: + href = dict(attrs).get("href") or "" + m = _ROW_HREF_RE.match(href) + if m: + self._row_match = ( + urllib.parse.unquote(m.group("repo")), + urllib.parse.unquote(m.group("provider")), + ) + elif tag == "td": + self._in_td = True + self._td_text = [] + + def handle_data(self, data: str) -> None: + if self._in_td: + self._td_text.append(data) + + def handle_endtag(self, tag: str) -> None: + if tag == "td": + self._in_td = False + text = "".join(self._td_text).strip() + if _PRICE_RE.match(text): + self._row_prices.append(float(text.replace("$", "").replace(",", ""))) + elif tag == "tr": + self._in_tr = False + if self._row_match and len(self._row_prices) >= 2: + repo_id, provider = self._row_match + self.quotes.append( + HfPriceQuote( + repo_id=repo_id, + provider=provider, + input_per_1m=self._row_prices[0], + output_per_1m=self._row_prices[1], + ) + ) + self._row_match = None + self._row_prices = [] + + +def parse_hf_pricing_table(html: str) -> list[HfPriceQuote]: + """Pure parsing function — no network I/O, so it's directly unit-testable.""" + parser = _HfPricingTableParser() + parser.feed(html) + return parser.quotes + + +def _default_fetch_html(url: str, *, timeout: float) -> str: + req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"}) + with urllib.request.urlopen(req, timeout=timeout) as resp: + return resp.read().decode("utf-8", errors="replace") + + +def fetch_hf_price_quotes( + search_term: str, + *, + fetch_html: Callable[[str], str] | None = None, + timeout: float = 15.0, +) -> list[HfPriceQuote]: + """Fetch and parse the HF inference pricing table filtered by ``search_term``. + + ``fetch_html`` is the test injection point (mirrors the ``backend=`` + convention used elsewhere in this package) — it takes the full URL and + returns the raw HTML text, so tests never hit the network. + """ + url = f"{HF_INFERENCE_MODELS_URL}?{urllib.parse.urlencode({'search': search_term})}" + if fetch_html is not None: + html = fetch_html(url) + else: + html = _default_fetch_html(url, timeout=timeout) + return parse_hf_pricing_table(html) + + +def cheapest_matching_quote( + quotes: list[HfPriceQuote], aliases: list[str] +) -> HfPriceQuote | None: + """Cheapest quote whose repo (optionally ``repo::provider``) is in ``aliases``. + + An alias of ``"org/repo"`` matches that repo under any provider; an + alias of ``"org/repo::provider"`` matches only that specific provider — + useful when only one provider's deployment has been human-verified as a + fair comparable (matching quantization/params). + """ + alias_set = {a.strip().lower() for a in aliases if isinstance(a, str) and a.strip()} + if not alias_set: + return None + matches = [q for q in quotes if alias_set & set(q.alias_keys())] + if not matches: + return None + return min(matches, key=lambda q: q.blended_price_per_1k_tokens()) + + +class HfPricingLog: + """Thread-safe SQLite-backed audit log of dynamic price changes (issue 23). + + Every price change (old, new, source alias/provider, timestamp) is + recorded here so a client dispute over a charge can be reconciled + against exactly what the market-tracking job did and when — mirrors + ``calibration.py``'s persistence shape. + """ + + def __init__(self, db_path: str | None = None) -> None: + self._db_path = db_path + self._lock = threading.Lock() + self._changes: list[dict] = [] + if self._db_path: + self._init_db() + self._load_from_db() + + def record_change( + self, + *, + model: str, + old_price_per_1k: float, + new_price_per_1k: float, + source_repo_id: str, + source_provider: str, + ts: float | None = None, + ) -> dict: + change = { + "model": model, + "old_price_per_1k": old_price_per_1k, + "new_price_per_1k": new_price_per_1k, + "source_repo_id": source_repo_id, + "source_provider": source_provider, + "ts": ts if ts is not None else time.time(), + } + with self._lock: + self._changes.append(change) + self._save_change(change) + return change + + def history(self, model: str | None = None, *, limit: int = 200) -> list[dict]: + with self._lock: + changes = list(self._changes) + if model is not None: + changes = [c for c in changes if c["model"] == model] + return changes[-limit:] + + # ---- persistence (billing.py / calibration.py pattern) ---- + + def _init_db(self) -> None: + con = sqlite3.connect(self._db_path) # type: ignore[arg-type] + con.execute( + "CREATE TABLE IF NOT EXISTS hf_price_changes " + "(id INTEGER PRIMARY KEY AUTOINCREMENT, model TEXT NOT NULL, " + "payload TEXT NOT NULL, ts REAL NOT NULL)" + ) + con.commit() + con.close() + + def _load_from_db(self) -> None: + con = sqlite3.connect(self._db_path) # type: ignore[arg-type] + rows = con.execute( + "SELECT payload FROM hf_price_changes ORDER BY ts, id" + ).fetchall() + con.close() + for (payload,) in rows: + try: + self._changes.append(json.loads(payload)) + except json.JSONDecodeError: + continue + + def _save_change(self, change: dict) -> None: + if not self._db_path: + return + con = sqlite3.connect(self._db_path) # type: ignore[arg-type] + con.execute( + "INSERT INTO hf_price_changes (model, payload, ts) VALUES (?, ?, ?)", + (change["model"], json.dumps(change), float(change["ts"])), + ) + con.commit() + con.close() + + +def hf_search_term(preset: dict, model_name: str) -> str: + """Best-effort search term for the HF pricing page's ``?search=`` filter.""" + hf_repo = preset.get("hf_repo") + if isinstance(hf_repo, str) and hf_repo: + return hf_repo.rsplit("/", 1)[-1] + return model_name + + +def refresh_preset_price( + *, + model_name: str, + preset: dict, + current_price: float, + fetch_html: Callable[[str], str] | None = None, + price_fraction: float = DEFAULT_CLIENT_PRICE_FRACTION, +) -> dict | None: + """Compute the new price for one preset, or None if nothing should change. + + Never raises — any fetch/parse failure or absence of a verified match is + treated identically: keep the static default (deliverable's fallback + requirement). Callers are responsible for actually applying the result + (``BillingLedger.set_price`` + logging), so this function stays a pure + "what should the new price be" computation and is trivially unit-testable. + """ + aliases = preset.get("hf_aliases") + if not aliases: + return None + try: + quotes = fetch_hf_price_quotes( + hf_search_term(preset, model_name), fetch_html=fetch_html + ) + quote = cheapest_matching_quote(quotes, aliases) + except Exception: + return None + if quote is None: + return None + new_price = round(quote.blended_price_per_1k_tokens() * price_fraction, 6) + if new_price <= 0: + return None + return { + "model": model_name, + "old_price_per_1k": current_price, + "new_price_per_1k": new_price, + "source_repo_id": quote.repo_id, + "source_provider": quote.provider, + } + + +__all__ = [ + "HF_INFERENCE_MODELS_URL", + "DEFAULT_HF_PRICING_LOG_DB_PATH", + "DEFAULT_CLIENT_PRICE_FRACTION", + "HfPriceQuote", + "HfPricingLog", + "parse_hf_pricing_table", + "fetch_hf_price_quotes", + "cheapest_matching_quote", + "hf_search_term", + "refresh_preset_price", +] diff --git a/packages/tracker/meshnet_tracker/model_presets.json b/packages/tracker/meshnet_tracker/model_presets.json index 4da4caa..7300186 100644 --- a/packages/tracker/meshnet_tracker/model_presets.json +++ b/packages/tracker/meshnet_tracker/model_presets.json @@ -11,6 +11,8 @@ ], "recommended": true, "deployment_status": "recommended", + "hf_aliases": [], + "hf_verified_match_note": "Pending human curation (issue 23) — no HF inference-marketplace listing has been confirmed as a comparable params/quantization match for this preset yet. Leave empty until a human signs off; an empty hf_aliases list keeps this model on the static default price.", "required_model_bytes": 638876385280, "download_size_bytes": 638876385280, "native_quantization": "int4", diff --git a/packages/tracker/meshnet_tracker/server.py b/packages/tracker/meshnet_tracker/server.py index 67d459f..62888b7 100644 --- a/packages/tracker/meshnet_tracker/server.py +++ b/packages/tracker/meshnet_tracker/server.py @@ -40,6 +40,8 @@ from .accounts import DEFAULT_ACCOUNTS_DB_PATH, AccountStore from .auth import is_validator_token, sign_hive_request, verify_hive_request from .wallet_proof import binding_message, verify_wallet_signature from .billing import DEFAULT_BILLING_DB_PATH, BillingLedger +from .calibration import DEFAULT_CALIBRATION_DB_PATH, ToplocCalibrationStore +from .hf_pricing import DEFAULT_HF_PRICING_LOG_DB_PATH, HfPricingLog, refresh_preset_price from .gossip import NodeGossip from .raft import RaftNode @@ -88,6 +90,14 @@ DEFAULT_MODEL_PRESETS: dict[str, dict] = { **_load_model_presets_from_data(), } +def _clone_model_presets(presets: dict[str, dict]) -> dict[str, dict]: + """Shallow-copy each preset dict so a runtime mutation (e.g. issue 23's + dynamic pricing refresh writing hf_last_price_per_1k/hf_last_updated) + never leaks into the shared module-level DEFAULT_MODEL_PRESETS and from + there into other TrackerServer instances in the same process.""" + return {name: dict(preset) for name, preset in presets.items()} + + DEFAULT_VRAM_BYTES = 8 * 1024 * 1024 * 1024 DEFAULT_RAM_BYTES = 16 * 1024 * 1024 * 1024 DEFAULT_QUANTIZATIONS = ["bfloat16"] @@ -976,6 +986,81 @@ def _nodes_and_bounds_for_model( return nodes, 0, max(node.num_layers for node in nodes) - 1 +def _fetch_toploc_commitment( + node: _NodeEntry, + *, + session_id: str, + model: str, + messages: list[dict], +) -> dict | None: + """Fetch a node's own on-demand TOPLOC boundary commitment (ADR-0018 §3), + same protocol as `ValidatorProcess._fetch_hop_commitment`.""" + endpoint = node.endpoint + if not isinstance(endpoint, str) or not endpoint: + return None + try: + req = urllib.request.Request( + f"{endpoint.rstrip('/')}/v1/audit/toploc/commitment", + data=json.dumps({ + "session_id": session_id, + "model": model, + "messages": messages, + "shard_start": node.shard_start, + "shard_end": node.shard_end, + }).encode(), + headers={"Content-Type": "application/json"}, + method="POST", + ) + with urllib.request.urlopen(req, timeout=5.0) as resp: + response = json.loads(resp.read()) + except (OSError, ValueError, json.JSONDecodeError): + return None + proof = response.get("toploc_proof") or response.get("activation_proof") + token_ids = response.get("claimed_token_ids") or response.get("output_token_ids") + if not isinstance(proof, dict): + return None + if not isinstance(token_ids, list) or not all(isinstance(t, int) for t in token_ids): + return None + return {"toploc_proof": proof, "claimed_token_ids": token_ids} + + +def _fetch_toploc_reference_activations( + reference_node_url: str, + *, + model: str, + messages: list[dict], + claimed_token_ids: list[int], + claim: Any, +) -> list | None: + """Teacher-force the claimed tokens on the reference node (same contract + as `ValidatorProcess._run_teacher_forced_prefill` / validator README's + "TOPLOC audit contract").""" + try: + req = urllib.request.Request( + f"{reference_node_url.rstrip('/')}/v1/audit/toploc", + data=json.dumps({ + "model": model, + "messages": messages, + "claimed_token_ids": claimed_token_ids, + "dtype": claim.dtype, + "quantization": claim.quantization, + "decode_batching_size": claim.decode_batching_size, + "topk": claim.topk, + "skip_prefill": claim.skip_prefill, + }).encode(), + headers={"Content-Type": "application/json"}, + method="POST", + ) + with urllib.request.urlopen(req, timeout=300.0) as resp: + response = json.loads(resp.read()) + except (OSError, ValueError, json.JSONDecodeError): + return None + activations = response.get("activations") + if not isinstance(activations, list): + return None + return activations + + def _load_directive(node: _NodeEntry, model: str, start: int, end: int, quantization: str) -> dict: return { "action": "LOAD_SHARD", @@ -1422,6 +1507,11 @@ 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, + toploc_calibration: "ToplocCalibrationStore | None" = None, + toploc_reference_node_url: str | None = None, + toploc_calibration_gate_min_hardware_profiles: int = 1, + toploc_backend: Any | None = None, + hf_pricing_log: "HfPricingLog | None" = None, ) -> None: super().__init__(addr, handler) self.registry = registry @@ -1443,6 +1533,13 @@ 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.toploc_calibration: ToplocCalibrationStore | None = toploc_calibration + self.toploc_reference_node_url = ( + toploc_reference_node_url.rstrip("/") if toploc_reference_node_url else None + ) + self.toploc_calibration_gate_min_hardware_profiles = toploc_calibration_gate_min_hardware_profiles + self.toploc_backend = toploc_backend + self.hf_pricing_log: HfPricingLog | None = hf_pricing_log class _TrackerHandler(http.server.BaseHTTPRequestHandler): @@ -1584,6 +1681,9 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): if self.path == "/v1/benchmark/hop-penalty": self._handle_benchmark_hop_penalty() return + if self.path == "/v1/calibration/toploc/run": + self._handle_toploc_calibration_run() + return if self.path == "/v1/wallet/register": self._handle_wallet_register() return @@ -1632,6 +1732,10 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): self._handle_admin_accounts() elif parsed.path == "/v1/benchmark/results": self._handle_benchmark_results() + elif parsed.path == "/v1/calibration/toploc/results": + self._handle_toploc_calibration_results() + elif parsed.path == "/v1/pricing/hf/history": + self._handle_hf_pricing_history(parsed) elif parsed.path == "/v1/registry/wallets": self._handle_registry_wallets() elif parsed.path in ("/dashboard", "/dashboard/"): @@ -3103,6 +3207,206 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): results = [] self._send_json(200, {"results": results if isinstance(results, list) else []}) + def _handle_toploc_calibration_run(self): + """Privileged: honest-noise TOPLOC calibration dispatch (issue 21). + + Fans the same fixed prompt through every currently registered node + that can solo-serve the full model (one pinned-route hop, mirroring + `_handle_benchmark_hop_penalty`'s dispatch pattern), then verifies + each node's own on-demand TOPLOC commitment against a teacher-forced + replay on the reference node — same audit contract the validator + uses (`packages/validator/README.md` "TOPLOC audit contract"). Each + node's raw divergence (not just pass/fail) is recorded into the + calibration corpus, keyed by wallet + GPU model + dtype, so + thresholds can eventually be derived instead of guessed. + + Nodes that only hold a partial shard (need a multi-hop route) are + skipped for this pass — solo dispatch isolates one node's hardware + noise without a route composition confound — and nodes that don't + answer the on-demand commitment fetch (endpoint down, or node-side + TOPLOC serving not yet wired) are skipped and reported, not treated + as a pass or a fail. + """ + server: _TrackerHTTPServer = self.server # type: ignore[assignment] + if not self._require_role("admin", "validator"): + return + if server.toploc_calibration is None: + self._send_json(503, {"error": "toploc calibration store is not enabled on this tracker"}) + return + if not server.toploc_reference_node_url: + self._send_json(503, {"error": "toploc_reference_node_url is not configured on this tracker"}) + return + auth = self.headers.get("Authorization") + body = self._read_json_body() + if body is None: + return + model = body.get("model", "") + if not model: + self._send_json(400, {"error": "model is required"}) + return + prompt = body.get("prompt") or "Calibration: say OK." + max_new_tokens = int(body.get("max_new_tokens", 32)) + seed = body.get("seed", 0) + + with server.lock: + self._purge_expired_nodes() + resolved = _nodes_and_bounds_for_model(server, model) + if resolved is None or not resolved[0]: + self._send_json(404, {"error": f"no nodes registered for model {model!r}"}) + return + all_nodes, rs, re = resolved + if server.contracts is not None: + all_nodes = [ + node for node in all_nodes + if not node.wallet_address or not server.contracts.registry.get_wallet(node.wallet_address).banned + ] + solo_nodes = [ + node for node in all_nodes + if node.shard_start is not None and node.shard_end is not None + and node.shard_start <= rs and node.shard_end >= re + ] + + self_url = f"http://127.0.0.1:{self.server.server_address[1]}" + messages = [{"role": "user", "content": prompt}] + node_results: list[dict] = [] + for node in solo_nodes: + request_id = f"cal-{uuid.uuid4().hex}" + request_body = json.dumps({ + "id": request_id, + "model": model, + "messages": messages, + "max_tokens": max_new_tokens, + "temperature": 0, + "seed": seed, + "route": [node.node_id], + }).encode() + req = urllib.request.Request( + f"{self_url}/v1/chat/completions", + data=request_body, + headers={"Content-Type": "application/json", "Authorization": auth}, + method="POST", + ) + try: + with urllib.request.urlopen(req, timeout=300.0) as resp: + json.loads(resp.read()) + except Exception as exc: + node_results.append({"node_id": node.node_id, "wallet_address": node.wallet_address, "error": str(exc)}) + continue + + outcome = self._verify_node_toploc_calibration( + server, node, request_id=request_id, model=model, messages=messages, + ) + node_results.append(outcome) + + skipped_partial_shard = [ + node.node_id for node in all_nodes if node not in solo_nodes + ] + record = { + "timestamp": time.time(), + "model": model, + "prompt_hash": hashlib.sha256(prompt.encode()).hexdigest()[:16], + "nodes": node_results, + "skipped_partial_shard_node_ids": skipped_partial_shard, + "gate_status": server.toploc_calibration.gate_status( + min_hardware_profiles=self._toploc_calibration_gate_min_hardware_profiles(), + ), + } + self._send_json(200, record) + + def _toploc_calibration_gate_min_hardware_profiles(self) -> int: + server: _TrackerHTTPServer = self.server # type: ignore[assignment] + return server.toploc_calibration_gate_min_hardware_profiles + + def _verify_node_toploc_calibration( + self, + server: "_TrackerHTTPServer", + node: "_NodeEntry", + *, + request_id: str, + model: str, + messages: list[dict], + ) -> dict: + """One node's calibration outcome: fetch its on-demand commitment, + teacher-force the claimed tokens on the reference node, verify, and + persist the raw divergence into the corpus.""" + from meshnet_validator.audit import ToplocProofClaim, verify_activation_proofs_detailed + + gpu_model = (node.hardware_profile or {}).get("gpu_name") or (node.hardware_profile or {}).get("device") or "unknown" + dtype = node.quantization or "unknown" + base_result = { + "node_id": node.node_id, + "wallet_address": node.wallet_address, + "gpu_model": gpu_model, + "dtype": dtype, + } + commitment = _fetch_toploc_commitment( + node, session_id=request_id, model=model, messages=messages, + ) + if commitment is None: + return {**base_result, "skipped": "no on-demand toploc commitment available"} + + try: + claim = ToplocProofClaim.from_mapping(commitment["toploc_proof"]) + except (KeyError, TypeError, ValueError): + return {**base_result, "skipped": "malformed toploc commitment"} + + reference_activations = _fetch_toploc_reference_activations( + server.toploc_reference_node_url, + model=model, + messages=messages, + claimed_token_ids=commitment["claimed_token_ids"], + claim=claim, + ) + if reference_activations is None: + return {**base_result, "skipped": "reference node teacher-forced replay failed"} + + result = verify_activation_proofs_detailed(reference_activations, claim, backend=server.toploc_backend) + if node.wallet_address: + server.toploc_calibration.record_run( + node_wallet=node.wallet_address, + gpu_model=gpu_model, + dtype=dtype, + model=model, + passed=result.passed, + exp_intersections=result.exp_intersections, + mant_err_mean=result.mant_err_mean, + mant_err_median=result.mant_err_median, + ) + return { + **base_result, + "passed": result.passed, + "exp_intersections": result.exp_intersections, + "mant_err_mean": result.mant_err_mean, + "mant_err_median": result.mant_err_median, + "chunk_count": result.chunk_count, + } + + def _handle_toploc_calibration_results(self): + server: _TrackerHTTPServer = self.server # type: ignore[assignment] + if not self._require_role("admin", "validator"): + return + if server.toploc_calibration is None: + self._send_json(503, {"error": "toploc calibration store is not enabled on this tracker"}) + return + min_profiles = self._toploc_calibration_gate_min_hardware_profiles() + self._send_json(200, { + "runs": server.toploc_calibration.runs(), + "envelope": server.toploc_calibration.envelope(), + "gate_status": server.toploc_calibration.gate_status(min_hardware_profiles=min_profiles), + }) + + def _handle_hf_pricing_history(self, parsed: urllib.parse.ParseResult): + """Dispute-auditability log for the dynamic HF-benchmarked pricing (issue 23).""" + server: _TrackerHTTPServer = self.server # type: ignore[assignment] + if not self._require_role("admin", "validator"): + return + if server.hf_pricing_log is None: + self._send_json(503, {"error": "hf pricing log is not enabled on this tracker"}) + return + params = urllib.parse.parse_qs(parsed.query) + model = params.get("model", [None])[0] + self._send_json(200, {"changes": server.hf_pricing_log.history(model)}) + def _handle_assign(self, parsed: urllib.parse.ParseResult): """Return an optimal shard assignment for a node given its hardware profile. @@ -3547,13 +3851,23 @@ class TrackerServer: validator_service_token: str | None = None, hive_secret: str | None = None, max_charge_per_request: float | None = None, + toploc_calibration: ToplocCalibrationStore | None = None, + toploc_calibration_db: str | None = None, + toploc_reference_node_url: str | None = None, + toploc_calibration_gate_min_hardware_profiles: int = 1, + toploc_backend: Any | None = None, + enable_hf_pricing: bool = False, + hf_pricing_log: HfPricingLog | None = None, + hf_pricing_log_db: str | None = None, + hf_pricing_refresh_interval: float = 86400.0, + hf_pricing_fetch_html: Any | None = None, ) -> None: 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) + model_presets if model_presets is not None else _clone_model_presets(DEFAULT_MODEL_PRESETS) ) self._contracts = contracts self._minimum_stake = minimum_stake @@ -3619,6 +3933,20 @@ class TrackerServer: if hive_secret is not None else os.environ.get("MESHNET_HIVE_SECRET") or None ) + if toploc_calibration is None and toploc_calibration_db: + toploc_calibration = ToplocCalibrationStore(db_path=toploc_calibration_db) + self._toploc_calibration: ToplocCalibrationStore | None = toploc_calibration + self._toploc_reference_node_url = toploc_reference_node_url + self._toploc_calibration_gate_min_hardware_profiles = toploc_calibration_gate_min_hardware_profiles + self._toploc_backend = toploc_backend + if hf_pricing_log is None and (enable_hf_pricing or hf_pricing_log_db): + hf_pricing_log = HfPricingLog(db_path=hf_pricing_log_db or DEFAULT_HF_PRICING_LOG_DB_PATH) + self._hf_pricing_log: HfPricingLog | None = hf_pricing_log + self._enable_hf_pricing = enable_hf_pricing + self._hf_pricing_refresh_interval = hf_pricing_refresh_interval + self._hf_pricing_fetch_html = hf_pricing_fetch_html + self._hf_pricing_stop = threading.Event() + self._hf_pricing_thread: threading.Thread | None = None self.port: int | None = None def start(self) -> int: @@ -3648,6 +3976,11 @@ class TrackerServer: validator_service_token=self._validator_service_token, hive_secret=self._hive_secret, max_charge_per_request=self._max_charge_per_request, + 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, + toploc_backend=self._toploc_backend, + hf_pricing_log=self._hf_pricing_log, ) self.port = self._server.server_address[1] @@ -3680,6 +4013,10 @@ class TrackerServer: self._settlement_stop.clear() self._settlement_thread = threading.Thread(target=self._settlement_loop, daemon=True) self._settlement_thread.start() + if self._enable_hf_pricing and self._billing is not None: + self._hf_pricing_stop.clear() + self._hf_pricing_thread = threading.Thread(target=self._hf_pricing_loop, daemon=True) + self._hf_pricing_thread.start() return self.port def _settlement_loop(self) -> None: @@ -3789,6 +4126,52 @@ class TrackerServer: flush=True, ) + def _hf_pricing_loop(self) -> None: + """Daily dynamic pricing refresh benchmarked against HF inference rates (issue 23). + + For every preset with a curated, human-verified ``hf_aliases`` list, + fetch current HF marketplace pricing and set the client price to 80% + of the cheapest matching alias. Presets with no (or an empty) + ``hf_aliases`` are left entirely alone — they keep the static + default price. Any single preset's fetch/parse failure is logged and + skipped; it never raises into this loop or the request path. + """ + billing = self._billing + assert billing is not None + while not self._hf_pricing_stop.wait(self._hf_pricing_refresh_interval): + for name, preset in list(self._model_presets.items()): + if not isinstance(preset, dict) or not preset.get("hf_aliases"): + continue + try: + result = refresh_preset_price( + model_name=name, + preset=preset, + current_price=billing.price_for(name), + fetch_html=self._hf_pricing_fetch_html, + ) + except Exception as exc: + print(f"[tracker] hf pricing refresh failed for {name!r}: {exc}", flush=True) + continue + if result is None: + continue + billing.set_price(name, result["new_price_per_1k"]) + preset["hf_last_price_per_1k"] = result["new_price_per_1k"] + preset["hf_last_updated"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) + if self._hf_pricing_log is not None: + self._hf_pricing_log.record_change( + model=name, + old_price_per_1k=result["old_price_per_1k"], + new_price_per_1k=result["new_price_per_1k"], + source_repo_id=result["source_repo_id"], + source_provider=result["source_provider"], + ) + print( + f"[tracker] hf pricing: {name} {result['old_price_per_1k']:.6f} -> " + f"{result['new_price_per_1k']:.6f} USDT/1k tokens " + f"(80% of {result['source_repo_id']}::{result['source_provider']})", + flush=True, + ) + def _raft_apply(self, command: str, payload: dict) -> None: """Called by RaftNode when a log entry is committed — replicate to local registry.""" if command != "register": @@ -3905,6 +4288,7 @@ class TrackerServer: self._stats_stop.set() self._deposit_stop.set() self._settlement_stop.set() + self._hf_pricing_stop.set() if self._stats is not None: self._stats.save_to_db() if self._billing is not None: @@ -3928,6 +4312,9 @@ class TrackerServer: if self._settlement_thread is not None: self._settlement_thread.join(timeout=1) self._settlement_thread = None + if self._hf_pricing_thread is not None: + self._hf_pricing_thread.join(timeout=1) + self._hf_pricing_thread = None self._server = None self._thread = None self._rebalance_thread = None diff --git a/packages/validator/README.md b/packages/validator/README.md index 9544f3d..e8c6ba3 100644 --- a/packages/validator/README.md +++ b/packages/validator/README.md @@ -51,8 +51,31 @@ skip_prefill = true encoding = "base64" ``` +`verify_activation_proofs_detailed()` (`meshnet_validator.audit`) surfaces the +raw TOPLOC divergence — `exp_intersections` (worst-case across chunks), +`mant_err_mean`, `mant_err_median` — alongside the pass/fail bool. This is +what the calibration corpus below is built from; existing callers that only +need the bool keep using `verify_activation_proofs()`. + +**Do not enable production audit thresholds before issue 21 closes.** Production audit thresholds remain gated on the honest-noise calibration -corpus in issue 21. +corpus in issue 21: the tracker's `POST /v1/calibration/toploc/run` +(admin/validator-only, mirrors `POST /v1/benchmark/hop-penalty`) dispatches a +fixed prompt to every solo-capable registered node, verifies each node's +on-demand commitment against a teacher-forced reference replay, and records +the divergence into a SQLite corpus (`meshnet_tracker.calibration. +ToplocCalibrationStore`) keyed by node wallet + GPU model + dtype. +`GET /v1/calibration/toploc/results` reports the corpus plus: + +- `envelope`: p99 honest-noise value per metric with a 20% safety margin — + the recommended (not yet wired) tolerance constants. +- `gate_status.ready`: whether the corpus covers enough distinct hardware + profiles (`--toploc-calibration-gate-min-hardware-profiles`, default 1). + **Alpha exception:** with the hired-VPS-only launch fleet, `ready` may + legitimately mean "covers every node we currently operate" — this must be + revisited (raise the minimum) before a public/volunteer launch broadens + the hardware mix, since a new corpus is required whenever the fleet's + hardware composition changes. Two operational notes: diff --git a/packages/validator/meshnet_validator/__init__.py b/packages/validator/meshnet_validator/__init__.py index 94b24a2..bf6c96b 100644 --- a/packages/validator/meshnet_validator/__init__.py +++ b/packages/validator/meshnet_validator/__init__.py @@ -8,7 +8,13 @@ import time import urllib.request from typing import Any -from .audit import ToplocAuditConfig, ToplocProofClaim, verify_activation_proofs +from .audit import ( + ToplocAuditConfig, + ToplocProofClaim, + ToplocVerificationResult, + verify_activation_proofs, + verify_activation_proofs_detailed, +) from .sampling import AdaptiveAuditSampler, AuditRateConfig from .tripwire import detect_output_tripwire diff --git a/packages/validator/meshnet_validator/audit.py b/packages/validator/meshnet_validator/audit.py index b870347..bda8058 100644 --- a/packages/validator/meshnet_validator/audit.py +++ b/packages/validator/meshnet_validator/audit.py @@ -91,6 +91,28 @@ def build_activation_proofs( ) +@dataclass(frozen=True) +class ToplocVerificationResult: + """Verification outcome plus the raw TOPLOC divergence metric. + + The `toploc` library's `verify_proofs_*` returns a bool for simple + prover/verifier config mismatches, but for a real activation comparison + it returns one `VerificationResult(exp_intersections, mant_err_mean, + mant_err_median)` per chunk (README §"What it actually is"). Historically + only `bool(result)` was kept, which is always true for a non-empty list + of results regardless of how divergent they are (AH-021 gap #1). This + dataclass surfaces the raw per-chunk metrics (aggregated: worst-case + `exp_intersections`, mean `mant_err_mean`/`mant_err_median`) so a + calibration corpus can be built before any threshold is trusted. + """ + + passed: bool + exp_intersections: float | None = None + mant_err_mean: float | None = None + mant_err_median: float | None = None + chunk_count: int = 0 + + def verify_activation_proofs( reference_activations: list[Any], claim: ToplocProofClaim, @@ -99,6 +121,25 @@ def verify_activation_proofs( backend: Any | None = None, ) -> bool: """Verify prover TOPLOC proofs against reference teacher-forced activations.""" + return verify_activation_proofs_detailed( + reference_activations, claim, config=config, backend=backend, + ).passed + + +def verify_activation_proofs_detailed( + reference_activations: list[Any], + claim: ToplocProofClaim, + *, + config: ToplocAuditConfig | None = None, + backend: Any | None = None, +) -> ToplocVerificationResult: + """Verify prover TOPLOC proofs and surface the raw divergence metric. + + Same pass/fail contract as `verify_activation_proofs` (kept as a thin + wrapper for existing call sites); this is the entry point for anything + that needs the underlying distance value, e.g. the AH-021 honest-noise + calibration corpus. + """ cfg = config or ToplocAuditConfig( dtype=claim.dtype, quantization=claim.quantization, @@ -108,23 +149,52 @@ def verify_activation_proofs( encoding=claim.encoding, ) if claim.dtype != cfg.dtype or claim.quantization != cfg.quantization: - return False + return ToplocVerificationResult(passed=False) if claim.decode_batching_size != cfg.decode_batching_size or claim.topk != cfg.topk: - return False + return ToplocVerificationResult(passed=False) if claim.skip_prefill != cfg.skip_prefill or claim.encoding != cfg.encoding: - return False + return ToplocVerificationResult(passed=False) module = backend or _load_toploc() function_name = f"verify_proofs_{claim.encoding}" verify = getattr(module, function_name) - return bool(_call_toploc( + raw = _call_toploc( verify, reference_activations, claim.proofs, decode_batching_size=claim.decode_batching_size, topk=claim.topk, skip_prefill=claim.skip_prefill, - )) + ) + divergence = _extract_divergence(raw) + return ToplocVerificationResult(passed=bool(raw), **divergence) + + +def _extract_divergence(raw: Any) -> dict[str, Any]: + """Aggregate per-chunk TOPLOC `VerificationResult`s, if present. + + `raw` is a plain bool for the simple fake backends used in existing unit + tests (no per-chunk metric available). The real `toploc` library returns + a list of per-chunk results; `exp_intersections` is aggregated by min + (worst honest-noise case across chunks) and the mantissa errors by mean. + """ + chunks = raw if isinstance(raw, (list, tuple)) else None + if not chunks: + return {"exp_intersections": None, "mant_err_mean": None, "mant_err_median": None, "chunk_count": 0} + exp_vals = [v for v in (_chunk_field(c, "exp_intersections") for c in chunks) if v is not None] + mean_vals = [v for v in (_chunk_field(c, "mant_err_mean") for c in chunks) if v is not None] + median_vals = [v for v in (_chunk_field(c, "mant_err_median") for c in chunks) if v is not None] + return { + "exp_intersections": min(exp_vals) if exp_vals else None, + "mant_err_mean": (sum(mean_vals) / len(mean_vals)) if mean_vals else None, + "mant_err_median": (sum(median_vals) / len(median_vals)) if median_vals else None, + "chunk_count": len(chunks), + } + + +def _chunk_field(chunk: Any, name: str) -> float | None: + value = chunk.get(name) if isinstance(chunk, dict) else getattr(chunk, name, None) + return float(value) if isinstance(value, (int, float)) else None def _load_toploc() -> Any: @@ -159,6 +229,8 @@ def _proof_encoding(value: object) -> ProofEncoding: __all__ = [ "ToplocAuditConfig", "ToplocProofClaim", + "ToplocVerificationResult", "build_activation_proofs", "verify_activation_proofs", + "verify_activation_proofs_detailed", ] diff --git a/tests/test_hf_pricing.py b/tests/test_hf_pricing.py new file mode 100644 index 0000000..5a2af99 --- /dev/null +++ b/tests/test_hf_pricing.py @@ -0,0 +1,150 @@ +"""AH-023: dynamic per-model pricing benchmarked against HF inference rates. + +Unit tests for the pure parsing/matching/computation pieces in +`meshnet_tracker.hf_pricing` — no network, no TrackerServer. The HTML +fixture below mirrors the row shape confirmed live against +https://huggingface.co/inference/models on 2026-07-06 (SSR'd table, one +`` anchor per +row, followed by `$input` / `$output` price cells). +""" + +from __future__ import annotations + +from meshnet_tracker.hf_pricing import ( + HfPriceQuote, + HfPricingLog, + cheapest_matching_quote, + parse_hf_pricing_table, + refresh_preset_price, +) + + +def _row(repo: str, provider: str, input_price: str, output_price: str) -> str: + return f""" + + + {repo} + link + + {provider} + ${input_price} + ${output_price} + 128000 + + """ + + +FIXTURE_HTML = f""" + + + {_row("zai-org/GLM-5.2", "novita", "1.40", "4.40")} + {_row("zai-org/GLM-5.2", "deepinfra", "0.93", "3.00")} + {_row("zai-org/GLM-4.7-Flash", "novita", "0.07", "0.40")} +
ModelProviderInput $/1MOutput $/1M
+""" + + +def test_parse_hf_pricing_table_extracts_repo_provider_and_prices(): + quotes = parse_hf_pricing_table(FIXTURE_HTML) + assert len(quotes) == 3 + assert quotes[0] == HfPriceQuote("zai-org/GLM-5.2", "novita", 1.40, 4.40) + assert quotes[1] == HfPriceQuote("zai-org/GLM-5.2", "deepinfra", 0.93, 3.00) + + +def test_blended_price_per_1k_tokens_is_average_of_input_output_over_1000(): + quote = HfPriceQuote("zai-org/GLM-5.2", "deepinfra", 0.93, 3.00) + assert quote.blended_price_per_1k_tokens() == (0.93 + 3.00) / 2 / 1000 + + +def test_cheapest_matching_quote_picks_lowest_blended_price_among_aliases(): + quotes = parse_hf_pricing_table(FIXTURE_HTML) + cheapest = cheapest_matching_quote(quotes, ["zai-org/GLM-5.2"]) + assert cheapest.provider == "deepinfra" + + +def test_cheapest_matching_quote_honors_repo_provider_scoped_alias(): + quotes = parse_hf_pricing_table(FIXTURE_HTML) + # Only the novita deployment was human-verified as comparable for this + # alias — the cheaper deepinfra row for the same repo must not match. + cheapest = cheapest_matching_quote(quotes, ["zai-org/GLM-5.2::novita"]) + assert cheapest.provider == "novita" + + +def test_cheapest_matching_quote_returns_none_when_no_alias_matches(): + quotes = parse_hf_pricing_table(FIXTURE_HTML) + assert cheapest_matching_quote(quotes, ["someone/unrelated-model"]) is None + + +def test_cheapest_matching_quote_returns_none_for_empty_aliases(): + quotes = parse_hf_pricing_table(FIXTURE_HTML) + assert cheapest_matching_quote(quotes, []) is None + + +def test_refresh_preset_price_end_to_end_with_injected_fetch(): + preset = {"hf_repo": "zai-org/GLM-5.2", "hf_aliases": ["zai-org/GLM-5.2"]} + result = refresh_preset_price( + model_name="glm-5.2", + preset=preset, + current_price=0.02, + fetch_html=lambda url: FIXTURE_HTML, + ) + assert result is not None + assert result["old_price_per_1k"] == 0.02 + # 80% of the cheapest matched (deepinfra) blended rate. + expected = round((0.93 + 3.00) / 2 / 1000 * 0.80, 6) + assert result["new_price_per_1k"] == expected + assert result["source_repo_id"] == "zai-org/GLM-5.2" + assert result["source_provider"] == "deepinfra" + + +def test_refresh_preset_price_skips_presets_without_hf_aliases(): + preset = {"hf_repo": "unsloth/Kimi-K2.7-Code"} + result = refresh_preset_price( + model_name="kimi-k2.7", + preset=preset, + current_price=0.02, + fetch_html=lambda url: FIXTURE_HTML, + ) + assert result is None + + +def test_refresh_preset_price_falls_back_silently_on_fetch_failure(): + preset = {"hf_repo": "zai-org/GLM-5.2", "hf_aliases": ["zai-org/GLM-5.2"]} + + def _boom(url: str) -> str: + raise ConnectionError("network unreachable") + + result = refresh_preset_price( + model_name="glm-5.2", preset=preset, current_price=0.02, fetch_html=_boom, + ) + assert result is None + + +def test_refresh_preset_price_falls_back_silently_when_no_match_found(): + preset = {"hf_repo": "zai-org/GLM-5.2", "hf_aliases": ["someone/unrelated-model"]} + result = refresh_preset_price( + model_name="glm-5.2", + preset=preset, + current_price=0.02, + fetch_html=lambda url: FIXTURE_HTML, + ) + assert result is None + + +def test_hf_pricing_log_persists_and_is_queryable(tmp_path): + db_path = str(tmp_path / "hf_pricing_log.sqlite") + log = HfPricingLog(db_path=db_path) + log.record_change( + model="glm-5.2", + old_price_per_1k=0.02, + new_price_per_1k=0.00157, + source_repo_id="zai-org/GLM-5.2", + source_provider="deepinfra", + ) + assert len(log.history()) == 1 + assert log.history("glm-5.2")[0]["new_price_per_1k"] == 0.00157 + assert log.history("some-other-model") == [] + + # Reopening against the same db path recovers the log (billing.py pattern). + reopened = HfPricingLog(db_path=db_path) + assert len(reopened.history()) == 1 diff --git a/tests/test_hf_pricing_dispatch.py b/tests/test_hf_pricing_dispatch.py new file mode 100644 index 0000000..64fcac7 --- /dev/null +++ b/tests/test_hf_pricing_dispatch.py @@ -0,0 +1,142 @@ +"""AH-023: end-to-end dynamic pricing refresh against a running TrackerServer. + +Verifies the full path the issue's acceptance criteria demand: a curated +alias -> injected "live" fetch -> 80%-of-cheapest computed price -> +BillingLedger.set_price() -> preset metadata updated -> change logged and +queryable via GET /v1/pricing/hf/history. Also verifies the required +fallback: a preset with no hf_aliases is left completely alone. +""" + +from __future__ import annotations + +import json +import time +import urllib.error +import urllib.request + +import pytest + +from meshnet_tracker.billing import BillingLedger +from meshnet_tracker.server import TrackerServer + +PRICED_MODEL = "zai-org/GLM-5.2" +STATIC_MODEL = "openai-community/gpt2" + + +def _row(repo: str, provider: str, input_price: str, output_price: str) -> str: + return f""" + + {repo} + link + + {provider} + ${input_price} + ${output_price} + + """ + + +FIXTURE_HTML = f""" + + {_row(PRICED_MODEL, "novita", "1.40", "4.40")} + {_row(PRICED_MODEL, "deepinfra", "0.93", "3.00")} +
+""" + + +def _get_json(url: str, headers: dict | None = None) -> dict: + req = urllib.request.Request(url, headers=headers or {}) + with urllib.request.urlopen(req) as r: + return json.loads(r.read()) + + +@pytest.fixture +def pricing_tracker(tmp_path): + ledger = BillingLedger(starting_credit=0.0, default_price_per_1k=0.02) + tracker = TrackerServer( + model_presets={ + PRICED_MODEL: { + "layers_start": 0, + "layers_end": 11, + "bytes_per_layer": {"bfloat16": 30 * 1024 * 1024}, + "hf_aliases": [PRICED_MODEL], + }, + STATIC_MODEL: { + "layers_start": 0, + "layers_end": 11, + "bytes_per_layer": {"bfloat16": 30 * 1024 * 1024}, + }, + }, + billing=ledger, + validator_service_token="pricing-token", + enable_hf_pricing=True, + hf_pricing_log_db=str(tmp_path / "hf_pricing_log.sqlite"), + hf_pricing_refresh_interval=0.05, + hf_pricing_fetch_html=lambda url: FIXTURE_HTML, + ) + port = tracker.start() + tracker_url = f"http://127.0.0.1:{port}" + yield tracker_url, ledger, tracker + tracker.stop() + + +def _wait_for_price_change(ledger: BillingLedger, model: str, *, timeout: float = 2.0) -> float: + deadline = time.time() + timeout + while time.time() < deadline: + price = ledger.price_for(model) + if price != 0.02: + return price + time.sleep(0.02) + raise AssertionError(f"price for {model!r} never changed from static default within {timeout}s") + + +def test_refresh_loop_repriced_model_with_curated_alias(pricing_tracker): + tracker_url, ledger, tracker = pricing_tracker + new_price = _wait_for_price_change(ledger, PRICED_MODEL) + expected = round((0.93 + 3.00) / 2 / 1000 * 0.80, 6) + assert new_price == expected + + preset = tracker._model_presets[PRICED_MODEL] + assert preset["hf_last_price_per_1k"] == expected + assert preset["hf_last_updated"] # ISO timestamp string, non-empty + + +def test_refresh_loop_leaves_model_without_hf_aliases_on_static_price(pricing_tracker): + tracker_url, ledger, tracker = pricing_tracker + _wait_for_price_change(ledger, PRICED_MODEL) # let the loop run at least once + assert ledger.price_for(STATIC_MODEL) == 0.02 + assert "hf_last_price_per_1k" not in tracker._model_presets[STATIC_MODEL] + + +def test_price_history_requires_auth(pricing_tracker): + tracker_url, ledger, tracker = pricing_tracker + _wait_for_price_change(ledger, PRICED_MODEL) + with pytest.raises(urllib.error.HTTPError) as exc_info: + _get_json(f"{tracker_url}/v1/pricing/hf/history") + assert exc_info.value.code == 401 + + +def test_price_history_reports_old_new_source_and_timestamp(pricing_tracker): + tracker_url, ledger, tracker = pricing_tracker + _wait_for_price_change(ledger, PRICED_MODEL) + result = _get_json( + f"{tracker_url}/v1/pricing/hf/history?model={PRICED_MODEL}", + headers={"Authorization": "Bearer pricing-token"}, + ) + assert len(result["changes"]) >= 1 + change = result["changes"][0] + assert change["old_price_per_1k"] == 0.02 + assert change["new_price_per_1k"] == round((0.93 + 3.00) / 2 / 1000 * 0.80, 6) + assert change["source_repo_id"] == PRICED_MODEL + assert change["source_provider"] == "deepinfra" + assert change["ts"] > 0 + + +def test_price_history_filters_by_model(pricing_tracker): + tracker_url, ledger, tracker = pricing_tracker + _wait_for_price_change(ledger, PRICED_MODEL) + result = _get_json( + f"{tracker_url}/v1/pricing/hf/history?model={STATIC_MODEL}", + headers={"Authorization": "Bearer pricing-token"}, + ) + assert result["changes"] == [] diff --git a/tests/test_toploc_audit.py b/tests/test_toploc_audit.py index 13c7757..2997d9d 100644 --- a/tests/test_toploc_audit.py +++ b/tests/test_toploc_audit.py @@ -2,10 +2,15 @@ from __future__ import annotations +from collections import namedtuple from types import SimpleNamespace from meshnet_validator import ToplocAuditConfig, ValidatorProcess -from meshnet_validator.audit import build_activation_proofs, verify_activation_proofs +from meshnet_validator.audit import ( + build_activation_proofs, + verify_activation_proofs, + verify_activation_proofs_detailed, +) class FakeToploc: @@ -199,3 +204,70 @@ def test_validator_rejects_swapped_precision_toploc_claim(): assert len(receipts) == 1 assert contracts.registry.slashes[0]["wallet_address"] == "wallet-bad" assert "TOPLOC activation proof mismatch" in contracts.registry.slashes[0]["reason"] + + +# AH-021: verify_activation_proofs_detailed surfaces the raw divergence +# metric a calibration corpus needs, instead of only a pass/fail bool. + +ChunkResult = namedtuple("ChunkResult", ["exp_intersections", "mant_err_mean", "mant_err_median"]) + + +class FakeToplocWithChunkResults: + """Mimics the real `toploc` library: verify returns per-chunk results, + not a bool, so `bool(result)` alone (the AH-021 gap #1 bug) is always + true for any non-empty response regardless of divergence.""" + + def build_proofs_base64(self, activations, *, decode_batching_size, topk, skip_prefill): + return {"activations": activations} + + def verify_proofs_base64(self, activations, proofs, *, decode_batching_size, topk, skip_prefill): + return [ + ChunkResult(exp_intersections=8, mant_err_mean=0.01, mant_err_median=0.008), + ChunkResult(exp_intersections=6, mant_err_mean=0.03, mant_err_median=0.02), + ] + + +def test_verify_activation_proofs_detailed_aggregates_per_chunk_divergence(): + fake_toploc = FakeToplocWithChunkResults() + activations = [[1.0, 2.0], [3.0, 4.0]] + config = ToplocAuditConfig(topk=2, decode_batching_size=16) + claim = build_activation_proofs(activations, config=config, backend=fake_toploc) + + result = verify_activation_proofs_detailed(activations, claim, config=config, backend=fake_toploc) + + assert result.passed is True # non-empty list is truthy, same as legacy behavior + assert result.chunk_count == 2 + assert result.exp_intersections == 6 # worst-case (min) across chunks + assert result.mant_err_mean == 0.02 # mean of per-chunk means + assert result.mant_err_median == 0.014 # mean of per-chunk medians + + # verify_activation_proofs still returns just the bool for existing callers. + assert verify_activation_proofs(activations, claim, config=config, backend=fake_toploc) is True + + +def test_verify_activation_proofs_detailed_no_metric_from_plain_bool_backend(): + fake_toploc = FakeToploc() + activations = [[1.0, 2.0, 3.0], [4.0, 5.0, 6.0]] + config = ToplocAuditConfig(topk=2, decode_batching_size=16) + claim = build_activation_proofs(activations, config=config, backend=fake_toploc) + + result = verify_activation_proofs_detailed(activations, claim, config=config, backend=fake_toploc) + + assert result.passed is True + assert result.chunk_count == 0 + assert result.exp_intersections is None + assert result.mant_err_mean is None + assert result.mant_err_median is None + + +def test_verify_activation_proofs_detailed_rejects_config_mismatch_without_calling_backend(): + fake_toploc = FakeToplocWithChunkResults() + activations = [[1.0, 2.0]] + canonical = ToplocAuditConfig(dtype="bfloat16", quantization="bfloat16", topk=2, decode_batching_size=16) + swapped = ToplocAuditConfig(dtype="bfloat16", quantization="int8", topk=2, decode_batching_size=16) + claim = build_activation_proofs(activations, config=swapped, backend=fake_toploc) + + result = verify_activation_proofs_detailed(activations, claim, config=canonical, backend=fake_toploc) + + assert result.passed is False + assert result.chunk_count == 0 diff --git a/tests/test_toploc_calibration.py b/tests/test_toploc_calibration.py new file mode 100644 index 0000000..c57ff6d --- /dev/null +++ b/tests/test_toploc_calibration.py @@ -0,0 +1,80 @@ +"""AH-021: honest-noise TOPLOC calibration corpus storage + aggregation.""" + +from __future__ import annotations + +from meshnet_tracker.calibration import ToplocCalibrationStore + + +def test_record_run_persists_and_reloads_from_sqlite(tmp_path): + db_path = str(tmp_path / "calibration.sqlite") + store = ToplocCalibrationStore(db_path=db_path) + store.record_run( + node_wallet="wallet-a", + gpu_model="RTX 4090", + dtype="bfloat16", + model="Qwen2.5-0.5B-Instruct", + passed=True, + exp_intersections=7.0, + mant_err_mean=0.01, + mant_err_median=0.008, + ) + + reloaded = ToplocCalibrationStore(db_path=db_path) + assert len(reloaded.runs()) == 1 + assert reloaded.runs()[0]["node_wallet"] == "wallet-a" + assert reloaded.distinct_hardware_profiles() == {("RTX 4090", "bfloat16")} + + +def test_gate_status_requires_minimum_distinct_hardware_profiles(): + store = ToplocCalibrationStore() + store.record_run( + node_wallet="wallet-a", gpu_model="RTX 4090", dtype="bfloat16", model="m", + passed=True, exp_intersections=8.0, mant_err_mean=0.01, mant_err_median=0.01, + ) + assert store.gate_status(min_hardware_profiles=2)["ready"] is False + + store.record_run( + node_wallet="wallet-b", gpu_model="A100", dtype="bfloat16", model="m", + passed=True, exp_intersections=8.0, mant_err_mean=0.01, mant_err_median=0.01, + ) + status = store.gate_status(min_hardware_profiles=2) + assert status["ready"] is True + assert status["distinct_hardware_profiles"] == 2 + + +def test_gate_status_empty_corpus_is_never_ready(): + store = ToplocCalibrationStore() + assert store.gate_status(min_hardware_profiles=0)["ready"] is False + + +def test_envelope_derives_thresholds_from_worst_case_percentile_with_margin(): + store = ToplocCalibrationStore() + # 100 honest runs; exp_intersections mostly 8, worst honest reading 5. + for i in range(100): + exp = 5.0 if i == 0 else 8.0 + mant = 0.05 if i == 0 else 0.01 + store.record_run( + node_wallet=f"wallet-{i}", gpu_model="RTX 4090", dtype="bfloat16", model="m", + passed=True, exp_intersections=exp, mant_err_mean=mant, mant_err_median=mant, + ) + + envelope = store.envelope(percentile=0.99, safety_margin=0.2) + assert envelope["sample_count"] == 100 + # p1 (tail) of exp_intersections is pulled down by the one worst honest + # reading (5.0 vs the 8.0 bulk); a 20% safety margin shaves it further. + assert envelope["recommended_min_exp_intersections"] < 8.0 + # p99 ceiling of mantissa error is likewise pulled up by the one worst + # honest reading (0.05 vs the 0.01 bulk), plus a 20% safety margin. + assert envelope["recommended_max_mant_err_mean"] > 0.01 + # In-sample FPR: at most the one outlier row should be flagged by its own + # derived thresholds. + assert 0.0 <= envelope["estimated_false_positive_rate"] <= 0.02 + + +def test_envelope_returns_none_when_no_samples(): + store = ToplocCalibrationStore() + envelope = store.envelope() + assert envelope["recommended_min_exp_intersections"] is None + assert envelope["recommended_max_mant_err_mean"] is None + assert envelope["recommended_max_mant_err_median"] is None + assert envelope["estimated_false_positive_rate"] is None diff --git a/tests/test_toploc_calibration_dispatch.py b/tests/test_toploc_calibration_dispatch.py new file mode 100644 index 0000000..cecf33f --- /dev/null +++ b/tests/test_toploc_calibration_dispatch.py @@ -0,0 +1,341 @@ +"""AH-021: tracker-scheduled TOPLOC honest-noise calibration dispatch. + +Extends the US-030 fleet-dispatch pattern (`_handle_benchmark_hop_penalty`) +from pinned-route latency benchmarking to a job that hits every solo-capable +registered node with a fixed prompt, verifies each node's own on-demand +TOPLOC commitment against a teacher-forced reference replay, and records the +raw divergence into the calibration corpus. +""" + +from __future__ import annotations + +import http.server +import json +import socketserver +import threading +import time +import urllib.error +import urllib.request + +import pytest + +from meshnet_tracker.server import TrackerServer +from meshnet_validator.audit import ToplocAuditConfig, build_activation_proofs + +MODEL = "openai-community/gpt2" +CONFIG = ToplocAuditConfig(topk=2, decode_batching_size=16, dtype="bfloat16", quantization="bfloat16") + + +class FakeToploc: + """Exact-equality fake backend, matching other TOPLOC test suites.""" + + def build_proofs_base64(self, activations, *, decode_batching_size, topk, skip_prefill): + return { + "activation_fingerprint": tuple(tuple(row) for row in activations), + "decode_batching_size": decode_batching_size, + "topk": topk, + "skip_prefill": skip_prefill, + } + + def verify_proofs_base64(self, activations, proofs, *, decode_batching_size, topk, skip_prefill): + # `proofs` may have round-tripped through JSON (HTTP dispatch), which + # turns the fingerprint's tuples into lists — normalize before compare. + fingerprint = proofs.get("activation_fingerprint") if isinstance(proofs, dict) else None + return ( + fingerprint is not None + and tuple(tuple(row) for row in fingerprint) == tuple(tuple(row) for row in activations) + and proofs.get("decode_batching_size") == decode_batching_size + and proofs.get("topk") == topk + and proofs.get("skip_prefill") == skip_prefill + ) + + +BACKEND = FakeToploc() + + +class FakeCalibrationNode: + """Stands in for a node: serves both /v1/chat/completions (tracker_mode + style) and its own on-demand TOPLOC commitment endpoint.""" + + def __init__(self, *, claim_activations, claimed_token_ids, response_text="ok", commitment_available=True): + self.requests: list[dict] = [] + claim = build_activation_proofs(claim_activations, config=CONFIG, backend=BACKEND) + outer = self + + class Handler(http.server.BaseHTTPRequestHandler): + def log_message(self, fmt, *args): + pass + + def do_POST(self): + length = int(self.headers.get("Content-Length", 0)) + body = json.loads(self.rfile.read(length) or b"{}") + if self.path == "/v1/chat/completions": + self._send_json(200, { + "id": "chatcmpl-cal", + "object": "chat.completion", + "created": int(time.time()), + "model": body.get("model", MODEL), + "choices": [{ + "index": 0, + "message": {"role": "assistant", "content": response_text}, + "finish_reason": "stop", + }], + "usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2}, + }) + return + if self.path == "/v1/audit/toploc/commitment": + outer.requests.append(body) + if not commitment_available: + self.send_response(404) + self.end_headers() + return + self._send_json(200, { + "toploc_proof": claim.as_mapping(), + "claimed_token_ids": claimed_token_ids, + }) + return + self.send_response(404) + self.end_headers() + + def _send_json(self, status, data): + 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) + + self._server = socketserver.ThreadingTCPServer(("127.0.0.1", 0), Handler) + self._server.daemon_threads = True + self._thread: threading.Thread | None = None + + def start(self) -> str: + self._thread = threading.Thread(target=self._server.serve_forever, daemon=True) + self._thread.start() + return f"http://127.0.0.1:{self._server.server_address[1]}" + + def stop(self) -> None: + self._server.shutdown() + self._server.server_close() + + +class FakeReferenceNode: + """Stands in for the reference node: teacher-forces the claimed tokens + and returns canned reference activations.""" + + def __init__(self, *, reference_activations): + self.requests: list[dict] = [] + outer = self + + class Handler(http.server.BaseHTTPRequestHandler): + def log_message(self, fmt, *args): + pass + + def do_POST(self): + if self.path != "/v1/audit/toploc": + self.send_response(404) + self.end_headers() + return + length = int(self.headers.get("Content-Length", 0)) + body = json.loads(self.rfile.read(length) or b"{}") + outer.requests.append(body) + payload = json.dumps({"activations": reference_activations}).encode() + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(payload))) + self.end_headers() + self.wfile.write(payload) + + self._server = socketserver.ThreadingTCPServer(("127.0.0.1", 0), Handler) + self._server.daemon_threads = True + self._thread: threading.Thread | None = None + + def start(self) -> str: + self._thread = threading.Thread(target=self._server.serve_forever, daemon=True) + self._thread.start() + return f"http://127.0.0.1:{self._server.server_address[1]}" + + def stop(self) -> None: + self._server.shutdown() + self._server.server_close() + + +def _post_json(url: str, payload: dict, headers: dict | None = None) -> dict: + req = urllib.request.Request( + url, + data=json.dumps(payload).encode(), + headers={"Content-Type": "application/json", **(headers or {})}, + method="POST", + ) + with urllib.request.urlopen(req) as r: + return json.loads(r.read()) + + +def _get_json(url: str, headers: dict | None = None) -> dict: + req = urllib.request.Request(url, headers=headers or {}) + with urllib.request.urlopen(req) as r: + return json.loads(r.read()) + + +@pytest.fixture +def calibration_setup(tmp_path): + reference_activations = [[1.0, 2.0], [3.0, 4.0]] + reference = FakeReferenceNode(reference_activations=reference_activations) + reference_url = reference.start() + + calibration_db = str(tmp_path / "calibration.sqlite") + tracker = TrackerServer( + model_presets={ + MODEL: {"layers_start": 0, "layers_end": 11, "bytes_per_layer": {"bfloat16": 30 * 1024 * 1024}}, + }, + validator_service_token="cal-token", + toploc_calibration_db=calibration_db, + toploc_reference_node_url=reference_url, + toploc_calibration_gate_min_hardware_profiles=1, + toploc_backend=BACKEND, + ) + port = tracker.start() + tracker_url = f"http://127.0.0.1:{port}" + + honest_node = FakeCalibrationNode( + claim_activations=reference_activations, # matches reference -> passes + claimed_token_ids=[101, 202], + ) + honest_node_url = honest_node.start() + reply = _post_json(f"{tracker_url}/v1/nodes/register", { + "endpoint": honest_node_url, + "shard_start": 0, + "shard_end": 11, + "model": MODEL, + "hardware_profile": {"gpu_name": "RTX 4090"}, + "quantization": "bfloat16", + "wallet_address": "wallet-honest", + "score": 1.0, + "tracker_mode": True, + }) + honest_node_id = reply["node_id"] + + partial_node = FakeCalibrationNode(claim_activations=reference_activations, claimed_token_ids=[101]) + partial_node_url = partial_node.start() + reply = _post_json(f"{tracker_url}/v1/nodes/register", { + "endpoint": partial_node_url, + "shard_start": 0, + "shard_end": 5, + "model": MODEL, + "hardware_profile": {"gpu_name": "A100"}, + "quantization": "bfloat16", + "wallet_address": "wallet-partial", + "score": 1.0, + "tracker_mode": True, + }) + partial_node_id = reply["node_id"] + + yield tracker_url, calibration_db, honest_node_id, partial_node_id + + honest_node.stop() + partial_node.stop() + reference.stop() + tracker.stop() + + +def test_calibration_run_requires_auth(calibration_setup): + tracker_url, _, _, _ = calibration_setup + with pytest.raises(urllib.error.HTTPError) as exc_info: + _post_json(f"{tracker_url}/v1/calibration/toploc/run", {"model": MODEL}) + assert exc_info.value.code == 401 + + +def test_calibration_run_dispatches_only_solo_capable_nodes(calibration_setup): + tracker_url, _, honest_node_id, partial_node_id = calibration_setup + record = _post_json( + f"{tracker_url}/v1/calibration/toploc/run", + {"model": MODEL, "prompt": "2+2", "max_new_tokens": 4}, + headers={"Authorization": "Bearer cal-token"}, + ) + assert record["skipped_partial_shard_node_ids"] == [partial_node_id] + assert len(record["nodes"]) == 1 + result = record["nodes"][0] + assert result["node_id"] == honest_node_id + assert result["wallet_address"] == "wallet-honest" + assert result["gpu_model"] == "RTX 4090" + assert result["dtype"] == "bfloat16" + assert result["passed"] is True + + +def test_calibration_run_persists_corpus_and_results_endpoint_reports_it(calibration_setup): + tracker_url, calibration_db, _, _ = calibration_setup + _post_json( + f"{tracker_url}/v1/calibration/toploc/run", + {"model": MODEL}, + headers={"Authorization": "Bearer cal-token"}, + ) + results = _get_json( + f"{tracker_url}/v1/calibration/toploc/results", + headers={"Authorization": "Bearer cal-token"}, + ) + assert len(results["runs"]) == 1 + assert results["runs"][0]["node_wallet"] == "wallet-honest" + assert results["gate_status"]["distinct_hardware_profiles"] == 1 + assert results["gate_status"]["ready"] is True + assert results["envelope"]["sample_count"] == 1 + + +def test_calibration_run_missing_reference_node_url_is_503(tmp_path): + tracker = TrackerServer( + model_presets={MODEL: {"layers_start": 0, "layers_end": 11}}, + validator_service_token="cal-token", + toploc_calibration_db=str(tmp_path / "calibration.sqlite"), + ) + port = tracker.start() + try: + with pytest.raises(urllib.error.HTTPError) as exc_info: + _post_json( + f"http://127.0.0.1:{port}/v1/calibration/toploc/run", + {"model": MODEL}, + headers={"Authorization": "Bearer cal-token"}, + ) + assert exc_info.value.code == 503 + finally: + tracker.stop() + + +def test_calibration_run_node_without_commitment_endpoint_is_skipped_not_failed(tmp_path): + reference = FakeReferenceNode(reference_activations=[[1.0, 2.0]]) + reference_url = reference.start() + tracker = TrackerServer( + model_presets={MODEL: {"layers_start": 0, "layers_end": 11}}, + validator_service_token="cal-token", + toploc_calibration_db=str(tmp_path / "calibration.sqlite"), + toploc_reference_node_url=reference_url, + ) + port = tracker.start() + tracker_url = f"http://127.0.0.1:{port}" + node = FakeCalibrationNode( + claim_activations=[[1.0, 2.0]], claimed_token_ids=[101], commitment_available=False, + ) + node_url = node.start() + _post_json(f"{tracker_url}/v1/nodes/register", { + "endpoint": node_url, + "shard_start": 0, + "shard_end": 11, + "model": MODEL, + "hardware_profile": {}, + "wallet_address": "wallet-no-commitment", + "score": 1.0, + "tracker_mode": True, + "node_id": "node-no-commitment", + }) + + try: + record = _post_json( + f"{tracker_url}/v1/calibration/toploc/run", + {"model": MODEL}, + headers={"Authorization": "Bearer cal-token"}, + ) + assert len(record["nodes"]) == 1 + assert "skipped" in record["nodes"][0] + assert record["gate_status"]["sample_count"] == 0 + finally: + node.stop() + reference.stop() + tracker.stop()