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

fix: reconcile legacy branch runtime with current GGUF

Browse Source
This commit is contained in:
Dobromir Popov
2026-07-17 14:21:02 +03:00
parent c758106a42
commit 6aced6a005
5 changed files with 89 additions and 325 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
packages/node/meshnet_node/capability.py
View File

@@ -20,8 +20,6 @@ import time
from dataclasses import dataclass, field
from typing import Any, Mapping
from .runtime_recipe import CompatibilityFingerprint, ShardIdentity
# Layout of the serialized report. Bump when the JSON shape changes.
CAPABILITY_SCHEMA_VERSION = 1
@@ -332,16 +330,7 @@ def _as_mapping(data: Any, field_name: str) -> Mapping[str, Any]:
@dataclass(frozen=True)
class CapabilityReport:
"""One node's validated (or failed) model/shard/recipe/backend combination.
`identity` is the exact DGR-003 artifact/runtime-recipe block: the separated
numerical axes and the compatibility fingerprint derived from them. It is
optional and additive — a node that predates DGR-003 presents none, and the
tracker falls back to the coarse label comparison it has always done
(ADR-0023's compat rollout). A node that *does* present one is held to it:
the tracker re-derives the fingerprint and refuses a report whose claim does
not match its own derivation.
"""
"""One node's validated (or failed) model/shard/recipe/backend combination."""
model: ModelIdentity
shard: ShardRange
@@ -352,7 +341,6 @@ class CapabilityReport:
duration_ms: int
diagnostics: tuple[str, ...] = ()
schema_version: int = CAPABILITY_SCHEMA_VERSION
identity: ShardIdentity | None = None
def __post_init__(self) -> None:
if self.status not in VALID_STATUSES:
@@ -372,11 +360,6 @@ class CapabilityReport:
def passed(self) -> bool:
return self.status == STATUS_PASSED
@property
def fingerprint(self) -> CompatibilityFingerprint | None:
"""The exact compatibility fingerprint, when this node declares one."""
return None if self.identity is None else self.identity.fingerprint
def identity_key(self) -> tuple[str, int, int, str, str, str, str]:
"""The tuple a consumer must match to reuse this proof.
@@ -397,7 +380,7 @@ class CapabilityReport:
return max(0.0, (time.time() if now is None else now) - self.validated_at)
def to_dict(self) -> dict:
doc = {
return {
"schema_version": self.schema_version,
"model": self.model.to_dict(),
"shard": self.shard.to_dict(),
@@ -408,9 +391,6 @@ class CapabilityReport:
"duration_ms": self.duration_ms,
"diagnostics": list(self.diagnostics),
}
if self.identity is not None:
doc["identity"] = self.identity.to_dict()
return doc
def to_json(self, indent: int | None = None) -> str:
return json.dumps(self.to_dict(), indent=indent, sort_keys=True)
@@ -437,7 +417,6 @@ class CapabilityReport:
):
raise CapabilityReportError("'validated_at' must be a Unix timestamp")
raw_identity = doc.get("identity")
return cls(
schema_version=schema_version,
model=ModelIdentity.from_dict(doc.get("model")),
@@ -448,9 +427,6 @@ class CapabilityReport:
validated_at=float(validated_at),
duration_ms=_require_int(doc.get("duration_ms"), "duration_ms", 0),
diagnostics=sanitize_diagnostics(doc.get("diagnostics")),
identity=(
None if raw_identity is None else ShardIdentity.from_dict(raw_identity)
),
)
@classmethod
@@ -485,14 +461,12 @@ def build_capability_report(
diagnostics: Any = None,
validated_at: float | None = None,
environ: Mapping[str, str] | None = None,
identity: ShardIdentity | None = None,
) -> CapabilityReport:
"""Assemble a report from flat validation results.
`model_config` may be the loaded config mapping (hashed into a fingerprint)
or an already-computed ``sha256:…`` string. `validated_at` defaults to now,
so callers that need determinism pass it explicitly. `identity` is the exact
DGR-003 artifact/recipe block, when the backend can state one.
so callers that need determinism pass it explicitly.
"""
return CapabilityReport(
model=ModelIdentity(
@@ -517,5 +491,4 @@ def build_capability_report(
validated_at=time.time() if validated_at is None else validated_at,
duration_ms=duration_ms,
diagnostics=sanitize_diagnostics(diagnostics, environ),
identity=identity,
)

5
packages/node/meshnet_node/torch_server.py
View File

@@ -19,6 +19,7 @@ from .model_backend import (
InsufficientVRAMError,
KVCacheMiss,
MissingModelDependencyError,
Quantization,
TailTokenResult,
TorchModelShard,
_tensor_from_bfloat16_bytes,
@@ -45,7 +46,7 @@ class _DirectRequestUncertainError(ConnectionError):
"""A direct request may have reached the downstream node but did not finish."""
from .server import ( # noqa: E402
from .server import (
_WIRE_VERSION,
_parse_shape,
_validate_activation_body,
@@ -398,7 +399,7 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler):
# Finite responses below provide Content-Length; streams are chunked.
protocol_version = "HTTP/1.1"
def log_message(self, fmt, *args): # suppress request logs in tests
def log_message(self, fmt, *args): # noqa: suppress request logs in tests
pass
def _request_id(self) -> str:

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

@@ -45,7 +45,7 @@ import urllib.parse
import urllib.request
import uuid
from collections import deque
from dataclasses import dataclass, field, replace
from dataclasses import dataclass, field
from importlib.resources import files
from pathlib import Path
from typing import Any
@@ -53,10 +53,13 @@ from typing import Any
from .accounts import DEFAULT_ACCOUNTS_DB_PATH, AccountStore
from .auth import is_validator_token, sign_hive_request, verify_hive_request
from .capability import (
DEFAULT_POLICY as DEFAULT_CAPABILITY_POLICY,
POLICY_COMPAT,
POLICY_ENFORCE,
STATE_ABSENT,
STATE_ADMITTED,
STATE_MODEL_MISMATCH,
STATE_SHARD_MISMATCH,
STATE_UNCERTIFIED,
CapabilityState,
absent_state,
evaluate_report,
@@ -65,7 +68,7 @@ from .capability import (
)
from .wallet_proof import binding_message, verify_wallet_signature
from .billing import DEFAULT_BILLING_DB_PATH, BillingLedger
from .calibration import ToplocCalibrationStore
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 .logging_setup import tracker_logger
@@ -81,13 +84,6 @@ from .routing_stats import (
)
from .model_files import files_for_layer_range, snapshot_dir_for_repo
from .raft import RaftNode
from .recipe import (
CertificationLedger,
DistributedForwardEvidence,
PresentedIdentity,
RecipeIdentityError,
RecipeStatus,
)
_CONSOLE_LIMIT = 1000
@@ -796,7 +792,6 @@ def _capability_from_registration(
hf_repo: str | None,
shard_start: int | None,
shard_end: int | None,
recipe_certifications: CertificationLedger,
) -> CapabilityState:
"""The tracker's verdict on the proof carried by one registration payload.
@@ -816,7 +811,6 @@ def _capability_from_registration(
declared_recipe_version=(
recipe_version if isinstance(recipe_version, str) else None
),
ledger=recipe_certifications,
)
@@ -836,11 +830,6 @@ def _admitted_nodes(nodes: list["_NodeEntry"], policy: str | None) -> list["_Nod
return [node for node in nodes if _capability_routable(node, effective)]
def _route_identity_partition(node: "_NodeEntry") -> tuple[str, ...]:
fingerprint = _node_admission(node).fingerprint
return ("legacy",) if fingerprint is None else ("exact", *fingerprint)
def _select_route(
nodes: list[_NodeEntry],
required_start: int,
@@ -864,51 +853,29 @@ def _select_route(
],
key=lambda n: (n.shard_start, -n.shard_end), # type: ignore[operator]
)
route: list[_NodeEntry] = []
covered_up_to = required_start - 1
def _routing_score(node: "_NodeEntry") -> float:
return _effective_throughput(node, model) * _reputation_multiplier(node, contracts)
partitions: dict[tuple[str, ...], list[_NodeEntry]] = {}
for node in candidates:
partitions.setdefault(_route_identity_partition(node), []).append(node)
while covered_up_to < required_end:
best: _NodeEntry | None = None
for node in candidates:
if node.shard_start <= covered_up_to + 1 and node.shard_end > covered_up_to:
if best is None:
best = node
elif node.shard_end > best.shard_end:
best = node
elif node.shard_end == best.shard_end and _routing_score(node) > _routing_score(best):
best = node
if best is None:
missing = covered_up_to + 1
return [], f"no route available: no registered node covers layer {missing}"
route.append(best)
covered_up_to = best.shard_end
candidates = [n for n in candidates if n is not best]
complete: list[list[_NodeEntry]] = []
furthest = required_start - 1
for partition in partitions.values():
pool = list(partition)
route: list[_NodeEntry] = []
covered_up_to = required_start - 1
while covered_up_to < required_end:
best: _NodeEntry | None = None
for node in pool:
if node.shard_start <= covered_up_to + 1 and node.shard_end > covered_up_to:
if best is None:
best = node
elif node.shard_end > best.shard_end:
best = node
elif (
node.shard_end == best.shard_end
and _routing_score(node) > _routing_score(best)
):
best = node
if best is None:
break
route.append(best)
covered_up_to = best.shard_end
pool = [node for node in pool if node is not best]
furthest = max(furthest, covered_up_to)
if covered_up_to >= required_end:
complete.append(route)
if not complete:
return [], f"no route available: no registered node covers layer {furthest + 1}"
route = max(
complete,
key=lambda candidate: (
min(_routing_score(node) for node in candidate),
-len(candidate),
),
)
return route, ""
@@ -944,12 +911,7 @@ def _enumerate_routes(
for head in heads:
route = [head]
covered_up_to = head.shard_end
head_partition = _route_identity_partition(head)
pool = [
node
for node in sharded
if node is not head and _route_identity_partition(node) == head_partition
]
pool = [n for n in sharded if n is not head]
while covered_up_to < required_end:
best = None
for n in pool:
@@ -2164,24 +2126,14 @@ def _find_pinned_route(
hop_count: int,
) -> list[_NodeEntry] | None:
"""First combination of exactly ``hop_count`` distinct nodes covering the
layer range, where every node extends coverage (US-030 benchmark routes).
Benchmark combos run real inference, so they obey the same DGR-003 rule as
every other route builder: one route, one exact identity. A combo that mixed
two fingerprints — or an exact Shard with a legacy one — would measure a
numerically incoherent route and record the garbage as a benchmark.
"""
layer range, where every node extends coverage (US-030 benchmark routes)."""
for combo in itertools.permutations(nodes, hop_count):
covered = required_start - 1
valid = True
partition = _route_identity_partition(combo[0])
for candidate in combo:
if candidate.shard_start is None or candidate.shard_end is None:
valid = False
break
if _route_identity_partition(candidate) != partition:
valid = False
break
if candidate.shard_start > covered + 1 or candidate.shard_end <= covered:
valid = False
break
@@ -2632,8 +2584,8 @@ def _estimate_prompt_tokens(body: dict) -> int | None:
def _requested_completion_token_limit(body: dict) -> int | None:
for key in ("max_completion_tokens", "max_tokens"):
value = body.get(key)
for field in ("max_completion_tokens", "max_tokens"):
value = body.get(field)
if isinstance(value, bool):
return None
if isinstance(value, (int, float)):
@@ -2862,42 +2814,6 @@ def _set_upstream_read_timeout(upstream: Any, timeout: float | None) -> None:
sock.settimeout(timeout)
def _upstream_wait_readable(upstream: Any, sock: Any, timeout: float) -> bool:
"""Return True when the next ``upstream.readline()`` can make progress.
Cancellation of a direct SSE proxy is polled with a bounded wait, but a
bounded socket timeout cannot be used for the read itself: ``socket``'s
makefile reader poisons itself after a single timeout, which would abort any
idle gap between frames. So the read stays blocking (timeout ``None``) and
this helper does the bounded waiting instead.
``select()`` alone is insufficient: ``urlopen``'s header parsing can read
the first SSE frame ahead into the response ``BufferedReader``, leaving the
raw socket with no new bytes -- ``select()`` would then withhold that frame
until the upstream produced more output. A non-blocking ``peek()`` surfaces
such already-buffered bytes first; only when nothing is buffered do we wait
on the raw socket.
"""
fp = getattr(upstream, "fp", None)
if fp is not None and sock is not None:
sock.setblocking(False)
try:
if fp.peek(1):
return True
except (BlockingIOError, InterruptedError):
pass
except (OSError, ValueError):
# Let the blocking readline surface the real error/EOF.
return True
finally:
# Restore blocking mode (timeout None) so readline never poisons.
sock.setblocking(True)
if sock is None:
return True
readable, _, _ = select.select([sock], [], [], timeout)
return bool(readable)
def _clear_proxy_progress_log_state(server: "_TrackerHTTPServer", request_id: str) -> None:
state = getattr(server, "_proxy_progress_log_state", None)
if state is not None:
@@ -3001,11 +2917,9 @@ class _TrackerHTTPServer(socketserver.ThreadingMixIn, http.server.HTTPServer):
relay_status: dict | None = None,
test_runner: "TestRunManager | None" = None,
capability_policy: str | None = None,
recipe_certifications: CertificationLedger | None = None,
) -> None:
super().__init__(addr, handler)
self.registry = registry
self.recipe_certifications = recipe_certifications or CertificationLedger()
self.capability_policy = normalize_policy(
capability_policy if capability_policy is not None else policy_from_env()
)
@@ -3050,7 +2964,7 @@ class _TrackerHTTPServer(socketserver.ThreadingMixIn, http.server.HTTPServer):
class _TrackerHandler(http.server.BaseHTTPRequestHandler):
def log_message(self, fmt, *args): # suppress request logs in tests
def log_message(self, fmt, *args): # noqa: suppress request logs in tests
pass
def _send_json(self, status: int, data: dict, headers: dict[str, str] | None = None) -> None:
@@ -3066,35 +2980,6 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
except BrokenPipeError:
pass
def _start_sse_response(self, status: int, content_type: str) -> tuple[Any, Any]:
"""Start a chunked SSE response and return its write/finish callbacks.
A streaming proxy cannot use EOF as its framing boundary: HTTP clients
and intermediaries are then free to retain an SSE frame until the
upstream closes. Chunked framing makes each flushed frame observable
immediately while still giving cancellation a clean response end.
"""
# The tracker otherwise serves finite HTTP/1.0-style responses. Limit
# HTTP/1.1 to this framed response so those existing response contracts
# do not become persistent connections accidentally.
self.protocol_version = "HTTP/1.1"
self.send_response(status)
self.send_header("Content-Type", content_type)
self.send_header("Cache-Control", "no-cache")
self.send_header("X-Accel-Buffering", "no")
self.send_header("Transfer-Encoding", "chunked")
self.end_headers()
def write(chunk: bytes) -> None:
self.wfile.write(f"{len(chunk):X}\r\n".encode() + chunk + b"\r\n")
self.wfile.flush()
def finish() -> None:
self.wfile.write(b"0\r\n\r\n")
self.wfile.flush()
return write, finish
# ---- unified auth boundary (ADR-0017) ----
def _resolve_identity(self) -> tuple[str | None, dict | None]:
@@ -4150,11 +4035,11 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
upstream = upstream_result[0]
with proxy_ctx.upstream_lock:
proxy_ctx.upstream = upstream
# Keep the read blocking (timeout None); cancellation is polled by
# _upstream_wait_readable() rather than by a bounded socket timeout,
# which would poison the makefile reader across idle SSE gaps.
_set_upstream_read_timeout(upstream, None)
upstream_sock = _upstream_socket(upstream)
if upstream_sock is not None:
_set_upstream_read_timeout(upstream, None)
else:
_set_upstream_read_timeout(upstream, 0.5)
_tracker_log(server, "info", "proxy connected", request_id=request_id, target_url=target_url)
except urllib.error.HTTPError as exc:
# Relay error status + body from node
@@ -4203,9 +4088,11 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
content_type = upstream.headers.get("Content-Type", "application/json")
if is_stream or "text/event-stream" in content_type:
# Relay SSE stream chunk-by-chunk
write_sse, finish_sse = self._start_sse_response(
200, "text/event-stream; charset=utf-8"
)
self.send_response(200)
self.send_header("Content-Type", "text/event-stream; charset=utf-8")
self.send_header("Cache-Control", "no-cache")
self.send_header("X-Accel-Buffering", "no")
self.end_headers()
stream_usage: dict | None = None
observed_stream_tokens = 0
client_gone = False
@@ -4213,14 +4100,22 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
while True:
if proxy_ctx.cancel_event.is_set():
break
if not _upstream_wait_readable(upstream, upstream_sock, 0.5):
if upstream_sock is not None:
readable, _, _ = select.select([upstream_sock], [], [], 0.5)
if not readable:
continue
try:
line = upstream.readline()
except TimeoutError:
continue
line = upstream.readline()
if not line:
if proxy_ctx.cancel_event.is_set():
break
break
if not client_gone:
try:
write_sse(line)
self.wfile.write(line)
self.wfile.flush()
except (BrokenPipeError, ConnectionResetError):
# Keep draining upstream so completed node work is still billed.
client_gone = True
@@ -4240,11 +4135,6 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
stream_usage = usage
except (BrokenPipeError, ConnectionResetError):
client_gone = True
except (OSError, ValueError):
# Cancellation closes the upstream from another handler
# thread to break the blocking readline safely.
if not proxy_ctx.cancel_event.is_set():
raise
if self._finalize_proxy_cancel(
proxy_ctx=proxy_ctx,
server=server,
@@ -4260,11 +4150,6 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
observed_stream_tokens=observed_stream_tokens,
stream_usage=stream_usage,
):
if not client_gone:
try:
finish_sse()
except (BrokenPipeError, ConnectionResetError):
pass
return
elapsed = time.monotonic() - started
# Bill even on client disconnect — the nodes did the work.
@@ -4297,11 +4182,6 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
api_key, model, tokens, node_work,
input_tokens=in_tokens, output_tokens=out_tokens,
)
if not client_gone:
try:
finish_sse()
except (BrokenPipeError, ConnectionResetError):
pass
else:
# Non-streaming: buffer and relay
resp_body = upstream.read()
@@ -4491,10 +4371,11 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
"""Forward a streamed relay response (US-036) to the client as SSE,
billing with the same accounting as the direct stream path."""
headers = first.get("headers") if isinstance(first.get("headers"), dict) else {}
write_sse, finish_sse = self._start_sse_response(
int(first.get("status", 200)),
headers.get("Content-Type", "text/event-stream; charset=utf-8"),
)
self.send_response(int(first.get("status", 200)))
self.send_header("Content-Type", headers.get("Content-Type", "text/event-stream; charset=utf-8"))
self.send_header("Cache-Control", "no-cache")
self.send_header("X-Accel-Buffering", "no")
self.end_headers()
server: _TrackerHTTPServer = self.server # type: ignore[assignment]
stream_usage: dict | None = None
observed_stream_tokens = 0
@@ -4508,7 +4389,8 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
data = chunk.encode()
if not client_gone:
try:
write_sse(data)
self.wfile.write(data)
self.wfile.flush()
except BrokenPipeError:
# Keep draining frames — the nodes did the work; bill it.
client_gone = True
@@ -4547,11 +4429,6 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
stream_usage=stream_usage,
)
):
if not client_gone:
try:
finish_sse()
except BrokenPipeError:
pass
return
elapsed = time.monotonic() - started
in_tokens, out_tokens = _stream_billable_split(
@@ -4582,11 +4459,6 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
api_key, model, tokens, node_work,
input_tokens=in_tokens, output_tokens=out_tokens,
)
if not client_gone:
try:
finish_sse()
except BrokenPipeError:
pass
def _send_relayed_response(self, response: dict) -> None:
status = int(response.get("status", 503))
@@ -4774,7 +4646,6 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
hf_repo=hf_repo,
shard_start=shard_start,
shard_end=shard_end,
recipe_certifications=server.recipe_certifications,
)
node_id = _node_id_for_registration(
@@ -6819,7 +6690,6 @@ class TrackerServer:
self._embedded_relay: Any | None = None
self._embedded_relay_actual_port: int | None = None
self._registry: dict[str, _NodeEntry] = {}
self._recipe_certifications = CertificationLedger()
self._lock = threading.Lock()
self._server: _TrackerHTTPServer | None = None
self._thread: threading.Thread | None = None
@@ -6929,84 +6799,6 @@ class TrackerServer:
self._test_runner: TestRunManager | None = test_runner
self.port: int | None = None
def certify_recipe(
self,
identity: PresentedIdentity,
evidence: DistributedForwardEvidence,
) -> RecipeStatus:
"""Promote one exact recipe from tracker-recorded distributed evidence.
This is the tracker's only certification authority. The evidence names
the nodes that served the forward; this method refuses to take any of
them on the evidence's word. For each one it goes back to what *this
tracker* re-derived at admission and requires the evidence to describe
the same Shard: same route fingerprint, same registered range, and the
same shard binding digest — the derivative's own bytes and exact range.
Without the binding check, a route fingerprint is range-independent by
design (:mod:`meshnet_tracker.recipe`), so evidence could name the right
recipe while describing splits nobody on the route was actually serving,
and the certification would attach to blobs that never ran.
"""
with self._lock:
for node_id, participant in zip(
evidence.node_ids, evidence.participants, strict=True
):
node = self._registry.get(node_id)
if node is None:
raise RecipeIdentityError(
f"certification participant {node_id!r} is not registered"
)
admitted = node.capability
if admitted.fingerprint != identity.key:
raise RecipeIdentityError(
f"certification participant {node_id!r} is not admitted under "
"the promoted fingerprint"
)
if admitted.state not in (STATE_UNCERTIFIED, STATE_ADMITTED):
raise RecipeIdentityError(
f"certification participant {node_id!r} has capability state "
f"{admitted.state!r}"
)
registered_range = (
node.shard_start,
None if node.shard_end is None else node.shard_end + 1,
)
if registered_range != (
participant.shard_start,
participant.shard_end,
):
raise RecipeIdentityError(
f"certification participant {node_id!r} range differs from its "
"registered Shard range"
)
# The exact bytes and range this node was admitted on. A node that
# registered without an identity has no binding, and cannot be a
# participant in an exact certification at all.
if admitted.shard_binding_digest is None:
raise RecipeIdentityError(
f"certification participant {node_id!r} registered no exact "
"Shard binding; it cannot certify a recipe"
)
if admitted.shard_binding_digest != participant.shard_binding_digest:
raise RecipeIdentityError(
f"certification participant {node_id!r} presents a Shard "
"binding this tracker did not admit it on; the evidence "
"describes different derivative bytes or a different range"
)
status = self._recipe_certifications.certify(identity, evidence)
for node in self._registry.values():
if (
node.capability.state == STATE_UNCERTIFIED
and node.capability.fingerprint == identity.key
):
node.capability = replace(
node.capability.with_state(STATE_ADMITTED, status.detail),
certification=status.status,
)
return status
def _start_embedded_relay(self) -> dict:
"""Start the shared RelayServer class in-process for tracker+relay deployments."""
if not self._embedded_relay_enabled:
@@ -7092,7 +6884,6 @@ class TrackerServer:
relay_status=http_relay_status,
test_runner=self._test_runner,
capability_policy=self._capability_policy,
recipe_certifications=self._recipe_certifications,
)
self.port = self._server.server_address[1]
@@ -7302,6 +7093,10 @@ class TrackerServer:
shard_end = int(payload["shard_end"]) if payload.get("shard_end") is not None else None
except (TypeError, ValueError):
return
try:
friendly_name = _normalize_friendly_name(payload.get("friendly_name"))
except ValueError:
friendly_name = None
# The replicated payload is the raw registration body, so the follower can
# resolve precision exactly as the leader did -- including telling a legacy
# absent `quantization` from a declared one. Dropping these fields here
@@ -7347,7 +7142,6 @@ class TrackerServer:
hf_repo=payload.get("hf_repo"),
shard_start=shard_start,
shard_end=shard_end,
recipe_certifications=self._recipe_certifications,
),
)
with self._lock: