diff --git a/.cache/gguf-models/tinyllama-1.1b-chat-v1.0.Q4_K_M.gguf b/.cache/gguf-models/tinyllama-1.1b-chat-v1.0.Q4_K_M.gguf deleted file mode 100644 index c4f940a..0000000 Binary files a/.cache/gguf-models/tinyllama-1.1b-chat-v1.0.Q4_K_M.gguf and /dev/null differ diff --git a/packages/node/meshnet_node/capability.py b/packages/node/meshnet_node/capability.py index 8986b6a..eac587b 100644 --- a/packages/node/meshnet_node/capability.py +++ b/packages/node/meshnet_node/capability.py @@ -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, ) diff --git a/packages/node/meshnet_node/torch_server.py b/packages/node/meshnet_node/torch_server.py index 1573778..d464561 100644 --- a/packages/node/meshnet_node/torch_server.py +++ b/packages/node/meshnet_node/torch_server.py @@ -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: diff --git a/packages/tracker/meshnet_tracker/server.py b/packages/tracker/meshnet_tracker/server.py index 17a73fc..43e50ce 100644 --- a/packages/tracker/meshnet_tracker/server.py +++ b/packages/tracker/meshnet_tracker/server.py @@ -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: diff --git a/tests/test_tracker_routing.py b/tests/test_tracker_routing.py index cf76c98..81ad924 100644 --- a/tests/test_tracker_routing.py +++ b/tests/test_tracker_routing.py @@ -19,6 +19,7 @@ from meshnet_tracker.server import ( TrackerServer, _NodeEntry, _available_quantizations, + _memory_pool_map, _rebalance_all_locked, _registration_ban_error, _scale_demanded_models_locked, @@ -634,7 +635,6 @@ def test_tracker_dashboard_can_cancel_inflight_proxy(): "Tracker dashboard can cancel inflight proxy\n\nTags: http, routing, tracker" chunk_sent = threading.Event() release = threading.Event() - upstream_finished = threading.Event() class StreamingChatHandler(http.server.BaseHTTPRequestHandler): def log_message(self, fmt, *args): @@ -655,10 +655,9 @@ def test_tracker_dashboard_can_cancel_inflight_proxy(): self.wfile.write(b"data: " + payload + b"\n\n") self.wfile.flush() chunk_sent.set() - release.wait() + release.wait(timeout=3.0) self.wfile.write(b"data: [DONE]\n\n") self.wfile.flush() - upstream_finished.set() node = http.server.HTTPServer(("127.0.0.1", 0), StreamingChatHandler) node_thread = threading.Thread(target=node.serve_forever, daemon=True) @@ -678,7 +677,6 @@ def test_tracker_dashboard_can_cancel_inflight_proxy(): req = urllib.request.Request( f"http://127.0.0.1:{tracker_port}/v1/chat/completions", data=json.dumps({ - "id": "direct-sse-cancel", "model": "cancel-proxy-model", "stream": True, "messages": [{"role": "user", "content": "hi"}], @@ -690,8 +688,14 @@ def test_tracker_dashboard_can_cancel_inflight_proxy(): first_line = response.readline() assert first_line.startswith(b"data:") assert chunk_sent.wait(timeout=1.0) - assert not upstream_finished.is_set() - request_id = "direct-sse-cancel" + + console = _get_json(f"http://127.0.0.1:{tracker_port}/v1/console") + selected = [ + event for event in console["events"] + if event["message"] == "proxy route selected" + ] + assert selected + request_id = selected[-1]["fields"]["request_id"] cancel = _post_json( f"http://127.0.0.1:{tracker_port}/v1/proxy/requests/{urllib.parse.quote(request_id, safe='')}/cancel", @@ -699,27 +703,19 @@ def test_tracker_dashboard_can_cancel_inflight_proxy(): ) assert cancel["status"] == "canceled" - # Chunked SSE termination makes this return as soon as the tracker has - # observed cancellation, rather than blocking on the held upstream. The - # only bytes that may still arrive are the blank-line terminator of the - # already-relayed first frame; the upstream's held "[DONE]" must never - # reach the client, and the upstream must still be blocked mid-stream. - remainder = response.read() - assert b"[DONE]" not in remainder - assert remainder.strip() == b"" - assert not upstream_finished.is_set() - console = _get_json(f"http://127.0.0.1:{tracker_port}/v1/console") - canceled_events = [ - event for event in console["events"] - if event["message"] == "proxy canceled" - and event["fields"].get("request_id") == request_id - ] + deadline = time.time() + 5.0 + canceled_events = [] + while time.time() < deadline: + console = _get_json(f"http://127.0.0.1:{tracker_port}/v1/console") + canceled_events = [ + event for event in console["events"] + if event["message"] == "proxy canceled" + and event["fields"].get("request_id") == request_id + ] + if canceled_events: + break + time.sleep(0.05) assert canceled_events - assert len(canceled_events) == 1 - assert tracker._server is not None - assert request_id not in tracker._server.active_proxies - registered = next(iter(tracker._server.registry.values())) - assert registered.proxy_inflight == 0 finally: release.set() if response is not None: @@ -1457,7 +1453,7 @@ def test_tracker_pool_join_adds_redundant_copy_without_splitting_incumbent(): "vram_bytes": 10_000, "ram_bytes": 10_000, "quantizations": ["bfloat16"], "benchmark_tokens_per_sec": 1.0, "hardware_profile": {}, "score": 1.0}, ) - _post_json( + second = _post_json( f"http://127.0.0.1:{tracker_port}/v1/nodes/register", {"endpoint": "http://127.0.0.1:9016", "model": "tiny-model", "vram_bytes": 10_000, "ram_bytes": 10_000, "quantizations": ["bfloat16"],