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

feat(tracker): add alpha calibration and dynamic pricing

Browse Source 
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.
This commit is contained in:
Dobromir Popov
2026-07-06 09:48:27 +03:00
parent 32514e84c9
commit f841dfaeed
18 changed files with 1996 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

389
packages/tracker/meshnet_tracker/server.py
View File

@@ -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