Merge remote-tracking branch 'origin/master' into ralph/proxy-stream-cancellation

2026-07-16 18:06:54 +03:00
parent 6fd9d93e4b 47b243cd98
commit 994546f78e
21 changed files with 923 additions and 99 deletions
--- a/tests/test_dashboard.py
+++ b/tests/test_dashboard.py
@@ -39,9 +39,14 @@ def test_dashboard_served_with_all_panels():
        assert "resolveModelGroup" in html
        assert "buildModelAliasMap" in html
        assert "modelAliasKey(raw)" in html
-        assert "main { display:grid; grid-template-columns:repeat(auto-fit,minmax(340px,1fr));" in html
+        assert "@media (min-width:900px)" in html
+        assert "grid-template-columns:repeat(4,minmax(0,1fr));" in html
+        assert ".wide { grid-column:span 2; }" in html
        assert 'onclick="clearConsole()"' in html
        assert "let consoleClearedAt = 0;" in html
+        assert "max-height:520px; overflow-y:auto; overflow-x:auto;" in html
+        assert "const CONSOLE_MAX_LINES = 1000;" in html
+        assert "events.slice(-CONSOLE_MAX_LINES)" in html
    finally:
        tracker.stop()

@@ -100,6 +105,39 @@ def test_dashboard_allows_admin_to_request_selected_model_load():
    assert '$("request-model-load").style.display = enabled ? "" : "none"' in html


+def test_dashboard_exposes_admin_model_inventory_and_release_controls():
+    "Admin placement controls show the full model inventory and can release capacity."
+    html = _dashboard_html()
+
+    assert 'id="admin-model-placement"' in html
+    assert "renderAdminModelPlacement" in html
+    assert '"/v1/models/release"' in html
+    assert "requestAdminModelLoad" in html
+    assert "releaseAdminModel" in html
+    assert 'data-admin-model-load=' in html
+    assert 'data-admin-model-release=' in html
+    assert "admin-model-placement-status" in html
+    assert 'id="admin-node-pool"' in html
+    assert "renderAdminNodePool" in html
+    assert "model drive" in html
+    # RAM and VRAM live in the network-map capacity object, not at node top level.
+    assert "node.ram_bytes = cap.ram_bytes" in html
+    assert "node.vram_bytes = cap.vram_bytes" in html
+    assert 'id="model-placement-dialog"' in html
+    assert "chooseModelPlacementNode" in html
+    assert "node_id: nodeId" in html
+    assert "modelAliasKey(node.model)" in html
+    assert 'id="model-placement-replace"' in html
+    assert 'id="model-placement-confirm"' in html
+    assert 'id="model-placement-replace-error"' in html
+    assert "force: replacing" in html
+    assert "Tick the box to confirm" in html
+    assert "releaseAllNodeModels" in html
+    assert '"/v1/nodes/release-all"' in html
+    assert "model RAM" in html
+    assert "loaded_model_bytes" in html
+
+
 def test_network_map_includes_node_friendly_name():
    "Network map includes node friendly name\n\nTags: dashboard, http"
    tracker = TrackerServer()
--- a/tests/test_dynamic_routing.py
+++ b/tests/test_dynamic_routing.py
@@ -355,6 +355,75 @@ def test_admin_model_load_request_queues_directive_on_joined_node():
    assert heartbeat["directives"][0]["model"] == "Qwen/Qwen2.5-0.5B-Instruct"


+def test_admin_can_replace_a_served_model_and_release_it():
+    "Forced admin placement replaces a served shard; release queues DROP_SHARD."
+    tracker = TrackerServer(enable_billing=False, validator_service_token="test-admin")
+    port = tracker.start()
+    try:
+        node = _post_json(
+            f"http://127.0.0.1:{port}/v1/nodes/register",
+            {"endpoint": "http://127.0.0.1:9912", "model": "stub-model",
+             "shard_start": 0, "shard_end": 3, "managed_assignment": True,
+             "max_loaded_shards": 1, "memory_mb": 1,
+             "hardware_profile": {"host_id": "full-host"}},
+        )
+        headers = {"Content-Type": "application/json", "Authorization": "Bearer test-admin"}
+        load = urllib.request.Request(
+            f"http://127.0.0.1:{port}/v1/models/load",
+            data=json.dumps({
+                "model": "qwen2.5-0.5b-instruct",
+                "node_id": node["node_id"],
+                "force": True,
+            }).encode(),
+            headers=headers, method="POST")
+        with urllib.request.urlopen(load) as response:
+            assert json.loads(response.read())["assignment"]["node_id"] == node["node_id"]
+        heartbeat = _post_json(f"http://127.0.0.1:{port}/v1/nodes/{node['node_id']}/heartbeat", {})
+        _post_json(
+            f"http://127.0.0.1:{port}/v1/nodes/{node['node_id']}/heartbeat",
+            {"completed_directives": [{"action": "DROP_SHARD", "model": "Qwen/Qwen2.5-0.5B-Instruct"}]},
+        )
+        network = _get_json(f"http://127.0.0.1:{port}/v1/network/map")
+        assert heartbeat["directives"][0]["action"] == "LOAD_SHARD"
+        release = urllib.request.Request(
+            f"http://127.0.0.1:{port}/v1/models/release",
+            data=json.dumps({"model": "qwen2.5-0.5b-instruct"}).encode(), headers=headers, method="POST")
+        with urllib.request.urlopen(release) as response:
+            assert json.loads(response.read())["released"] == 1
+        heartbeat = _post_json(f"http://127.0.0.1:{port}/v1/nodes/{node['node_id']}/heartbeat", {})
+    finally:
+        tracker.stop()
+
+    assert heartbeat["directives"][0]["action"] == "DROP_SHARD"
+    released_node = next(item for item in network["nodes"] if item["node_id"] == node["node_id"])
+    assert released_node["shard_start"] is None
+    assert released_node["shard_end"] is None
+
+
+def test_models_list_does_not_duplicate_a_preset_registered_by_hf_repo():
+    """A preset and its canonical repository are one selectable model."""
+    tracker = TrackerServer(enable_billing=False)
+    port = tracker.start()
+    try:
+        _post_json(
+            f"http://127.0.0.1:{port}/v1/nodes/register",
+            {
+                "endpoint": "http://127.0.0.1:9913",
+                "model": "Qwen2.5-0.5B-Instruct",
+                "hf_repo": "Qwen/Qwen2.5-0.5B-Instruct",
+                "num_layers": 24,
+                "shard_start": 0,
+                "shard_end": 23,
+            },
+        )
+        models = _get_json(f"http://127.0.0.1:{port}/v1/models")["data"]
+    finally:
+        tracker.stop()
+
+    assert [model["id"] for model in models].count("qwen2.5-0.5b-instruct") == 1
+    assert not any(model["id"] == "Qwen/Qwen2.5-0.5B-Instruct" for model in models)
+
+
 def test_endpoint_key_distinguishes_same_port_different_hosts():
    "Endpoint key distinguishes same port different hosts\n\nTags: http, performance, routing, tracker"
    from meshnet_node.torch_server import _clamp_downstream_hops, _endpoint_key
--- a/tests/test_node_admission.py
+++ b/tests/test_node_admission.py
@@ -358,6 +358,73 @@ def test_a_stale_report_cannot_be_reused_to_register(startup_env):
    assert startup_env == []


+# ---------------------------------------------------------------------------
+# Re-registration: the proof presented is fresh, never the one captured at boot
+# ---------------------------------------------------------------------------
+
+
+def test_run_startup_hands_the_heartbeat_a_refresher_for_the_current_shard(startup_env, monkeypatch):
+    "The tracker refuses aged proofs, so the heartbeat must be able to re-prove what the node serves now.\n\nTags: node, admission, startup"
+    import meshnet_node.startup as startup_mod
+
+    captured: dict = {}
+    monkeypatch.setattr(
+        startup_mod, "_start_heartbeat", lambda *a, **kw: captured.update(kw)
+    )
+
+    _start(capability_validator=capability_stub())
+
+    refresh = captured.get("refresh_capability")
+    assert callable(refresh), "run_startup no longer wires a capability refresher"
+    fresh = refresh({"hf_repo": MODEL, "model": MODEL.split("/")[-1]})
+    assert fresh is not None
+    assert fresh["model"]["model_id"] == MODEL
+    assert (fresh["shard"]["start"], fresh["shard"]["end"]) == (0, 23)
+    assert fresh["validated_at"] > time.time() - 60
+
+
+def test_a_reregistration_presents_a_refreshed_proof(monkeypatch):
+    "Replaying the boot-time report after an outage re-registers the node unroutable; the re-register path must present a fresh proof.\n\nTags: node, admission, startup"
+    import json
+
+    import meshnet_node.startup as startup_mod
+
+    model = "acme/refresh-model-7b"
+    boot_report = {"validated_at": 1.0, "marker": "boot"}
+    fresh_report = {"validated_at": 2.0, "marker": "fresh"}
+    posted: list[dict] = []
+
+    def _record(url, payload, timeout=10.0):
+        if url.endswith("/v1/nodes/register") and payload.get("hf_repo") == model:
+            posted.append(json.loads(json.dumps(payload)))
+            return {"node_id": "node-refresh"}
+        raise SystemExit  # first heartbeat POST ends the daemon loop
+
+    monkeypatch.setattr(startup_mod, "_post_json", _record)
+
+    payload = {
+        "hf_repo": model,
+        "model": model.split("/")[-1],
+        "capability_report": dict(boot_report),
+    }
+    thread = startup_mod._start_heartbeat(
+        "http://tracker.invalid",
+        startup_mod._PENDING_NODE_ID,  # forces a re-registration on the first tick
+        payload,
+        interval=0.02,
+        refresh_capability=lambda _payload: dict(fresh_report),
+    )
+
+    # The loop must be dead before this test returns: once monkeypatch restores
+    # `_post_json`, a surviving thread would re-register through whatever the
+    # *next* test patches in and corrupt its call counts.
+    thread.join(timeout=5.0)
+
+    assert not thread.is_alive(), "the heartbeat loop outlived the test"
+    assert posted, "the heartbeat never re-registered"
+    assert posted[0]["capability_report"] == fresh_report
+
+
 def test_a_matching_passing_report_registers_and_travels_with_the_payload(startup_env):
    "Registration carries the proof for exactly the model/shard/recipe it advertises.\n\nTags: node, admission, startup"
    node = _start()  # production validator against a working fake backend
--- a/tests/test_node_startup.py
+++ b/tests/test_node_startup.py
@@ -287,6 +287,47 @@ def test_configure_torch_threads_applies_explicit_settings(monkeypatch):
    assert active == {"torch_threads": 12, "torch_interop_threads": 2}


+def test_heartbeat_applies_release_without_reregistering(monkeypatch):
+    """DROP_SHARD has no replacement range and must not look like an outage."""
+    import meshnet_node.startup as startup_mod
+
+    released = threading.Event()
+    requests: list[tuple[str, dict]] = []
+
+    class FakeNode:
+        def apply_tracker_directives(self, directives):
+            assert directives == [{"action": "DROP_SHARD", "model": "Qwen/Qwen2.5-0.5B-Instruct"}]
+            return {"action": "DROP_SHARD", "model": "Qwen/Qwen2.5-0.5B-Instruct"}
+
+    def fake_post(url, payload, timeout=10.0):
+        requests.append((url, dict(payload)))
+        released.set()
+        return {"directives": [{"action": "DROP_SHARD", "model": "Qwen/Qwen2.5-0.5B-Instruct"}]}
+
+    sleep_calls = 0
+
+    def one_heartbeat(_seconds):
+        nonlocal sleep_calls
+        sleep_calls += 1
+        if sleep_calls > 1:
+            raise SystemExit
+
+    monkeypatch.setattr(startup_mod, "_post_json", fake_post)
+    monkeypatch.setattr(startup_mod.time, "sleep", one_heartbeat)
+    payload = {
+        "model": "Qwen2.5-0.5B-Instruct",
+        "hf_repo": "Qwen/Qwen2.5-0.5B-Instruct",
+        "shard_start": 0,
+        "shard_end": 23,
+    }
+    startup_mod._start_heartbeat("http://tracker", "node-1", payload, interval=0, node_ref=FakeNode())
+
+    assert released.wait(1), "heartbeat did not receive the queued release"
+    assert len(requests) == 1, "release must not trigger a re-registration"
+    assert payload["shard_start"] == 0
+    assert payload["shard_end"] == 23
+
+
 def test_benchmark_throughput_is_registered_in_payload(monkeypatch, tmp_path):
    "benchmark_tokens_per_sec from the benchmark is included in the tracker registration.\n\nTags: node, performance, startup"
    import meshnet_node.startup as startup_mod
--- a/tests/test_route_session_benchmark.py
+++ b/tests/test_route_session_benchmark.py
@@ -32,12 +32,18 @@ def test_matrix_reports_direct_relay_prefill_decode_and_machine_readable_metrics
        assert {"p50_latency_ms", "p95_latency_ms", "payload_bytes", "compression_ratio",
                "connection_attempts", "p95_queue_wait_ms"} <= set(run["phases"]["decode"])
        sample = run["samples"][0]
+        assert sample["model_ms"] > 0
+        assert sample["encode_ms"] > 0
+        assert sample["activation_decode_ms"] > 0
        assert sample["framing_ms"] > 0
        assert sample["metadata_ms"] > 0
        assert sample["copy_allocation_ms"] > 0
        assert sample["copy_allocation_bytes"] >= sample["payload_bytes"]
+        assert sample["local_http_forwarding_ms"] > 0
        assert len(run["samples"]) == 1 + len(run["output_tokens"])
-        assert {"tokens_per_sec", "bytes_per_token", "compression_cpu_ms", "peak_buffered_bytes"} <= set(run["phases"]["decode"])
+        assert {"tokens_per_sec", "bytes_per_token", "compression_cpu_ms", "peak_buffered_bytes",
+                "model_execution_ms", "activation_encoding_ms", "activation_decoding_ms",
+                "local_http_forwarding_ms"} <= set(run["phases"]["decode"])


 def test_cached_sessions_reuse_one_connection_and_preserve_stub_tokens():
@@ -74,7 +80,10 @@ def test_cli_writes_json_artifact_and_human_summary(tmp_path, capsys):
    report = json.loads(output.read_text())
    assert report["schema_version"] == 1
    assert "Route Session benchmark" in capsys.readouterr().out
-    assert "relay" in format_summary(report)
+    summary = format_summary(report)
+    assert "relay" in summary
+    assert "model/encode/decode" in summary
+    assert "HTTP" in summary


 def test_performance_gate_checks_comparison_identity_session_and_cleanup():
--- a/tests/test_tracker_routing.py
+++ b/tests/test_tracker_routing.py
@@ -2874,6 +2874,43 @@ def test_same_endpoint_can_register_multiple_models():
        tracker.stop()


+def test_explicit_model_placement_targets_only_the_selected_node():
+    "An admin can add and release a model on one chosen multi-model node.\n\nTags: http, routing, tracker"
+    from meshnet_tracker.server import _release_model_locked, _request_model_load_locked
+
+    tracker = _tracker(model_presets={
+        "model-a": {"total_layers": 4, "bytes_per_layer": {"bfloat16": 1_000}, "hf_repo": "org/ModelA"},
+        "model-b": {"total_layers": 4, "bytes_per_layer": {"bfloat16": 1_000}, "hf_repo": "org/ModelB"},
+    })
+    tracker_port = tracker.start()
+    try:
+        registrations = []
+        for port in (9062, 9063):
+            registrations.append(_post_json(
+                f"http://127.0.0.1:{tracker_port}/v1/nodes/register",
+                {"endpoint": f"http://127.0.0.1:{port}", "model": "model-a", "hf_repo": "org/ModelA",
+                 "num_layers": 4, "shard_start": 0, "shard_end": 3, "max_loaded_shards": 2,
+                 "vram_bytes": 20_000, "ram_bytes": 20_000, "quantizations": ["bfloat16"],
+                 "benchmark_tokens_per_sec": 1.0, "hardware_profile": {}, "score": 1.0},
+            ))
+        selected, other = (item["node_id"] for item in registrations)
+
+        with tracker._lock:
+            assignment = _request_model_load_locked(tracker._server, "model-b", selected)  # type: ignore[arg-type]
+        assert assignment is not None
+        assert assignment["node_id"] == selected
+        assert tracker._registry[selected].pending_new_assignment is not None
+        assert tracker._registry[other].pending_new_assignment is None
+
+        with tracker._lock:
+            released = _release_model_locked(tracker._server, "model-a", selected)  # type: ignore[arg-type]
+        assert released == 1
+        assert len(tracker._registry[selected].pending_directives) == 2
+        assert tracker._registry[other].pending_directives == []
+    finally:
+        tracker.stop()
+
+
 def test_scale_demanded_models_queues_add_shard_on_spare_host():
    "Scale demanded models queues add shard on spare host\n\nTags: http, routing, tracker"
    tracker = _tracker(model_presets={
@@ -3010,6 +3047,13 @@ def test_a_node_declaring_an_unsupported_quantization_is_never_routed():
    assert status == 503


+def test_a_node_declaring_auto_quantization_serves_a_default_precision_request():
+    "'auto' is the CLI default that delegates the choice — it is not a refusal, so the node must resolve to its best advertised precision and route.\n\nTags: http, routing, tracker"
+    status, response = _proxy_chat_status(POLICY_COMPAT, quantization="auto")
+    assert status == 200
+    assert response["choices"][0]["message"]["content"] == "ok"
+
+
 def test_a_node_declaring_a_null_quantization_is_never_routed():
    "An explicit null states 'no usable precision' -- only an absent field is legacy.\n\nTags: http, routing, tracker"
    node = http.server.HTTPServer(("127.0.0.1", 0), _EchoChatHandler)