From 040f014fa3d65c6fe9720752dc13e3a469963abd Mon Sep 17 00:00:00 2001 From: Will Ford Date: Mon, 7 Sep 2026 23:08:43 +0200 Subject: [PATCH 1/7] manual-nodes: adopt declared OpenAI-compatible endpoints and persist entries node/add accepts exactly one of address (historical default-port probing) or openai_base_url (http only). Declared endpoints are probed at their own URL: GET {base}/models doubles as liveness and model list, with best-effort node-info on the URL's host; the default engine-port legs are skipped. The service now owns its entry list in the app data directory (atomic tmp+rename) and restores it at startup, so manual nodes survive a restart. Signed-off-by: Will Ford --- services/nvpair-manual-nodes/manager.go | 171 +++++++++++++++++-- services/nvpair-manual-nodes/manager_test.go | 153 +++++++++++++++++ services/nvpair-manual-nodes/store.go | 110 ++++++++++++ services/nvpair-manual-nodes/store_test.go | 112 ++++++++++++ 4 files changed, 529 insertions(+), 17 deletions(-) create mode 100644 services/nvpair-manual-nodes/store.go create mode 100644 services/nvpair-manual-nodes/store_test.go diff --git a/services/nvpair-manual-nodes/manager.go b/services/nvpair-manual-nodes/manager.go index 55a4040a..d1a0510f 100644 --- a/services/nvpair-manual-nodes/manager.go +++ b/services/nvpair-manual-nodes/manager.go @@ -12,7 +12,9 @@ import ( "log/slog" "net" "net/http" + "net/url" "strconv" + "strings" "sync" "time" @@ -93,6 +95,11 @@ type ManualEntry struct { Name string `json:"name"` TLSPort int `json:"tls_port,omitempty"` MTLS bool `json:"mtls,omitempty"` + // OpenAIBaseURL declares an externally-managed OpenAI-compatible + // endpoint by its base URL (e.g. http://host:8888/v1) instead of + // probing a host for the default engine ports. Mutually exclusive + // with Address; exactly one of the two must be set. + OpenAIBaseURL string `json:"openai_base_url,omitempty"` } // ManualNodeStatus mirrors a manual entry plus the latest probe @@ -115,7 +122,19 @@ type ManualNodeStatus struct { LMStudioUp bool `json:"lmstudio_up"` LMStudioPort int `json:"lmstudio_port"` LMStudioModels []string `json:"lmstudio_models,omitempty"` - NodeInfoUp bool `json:"node_info_up"` + // The openai_* fields describe a declared OpenAI-compatible endpoint + // (an entry added with openai_base_url). OpenAIBaseURL echoes the + // declared URL so the UI shows what was configured; OpenAIHost/Port + // and OpenAIBasePath are the parsed parts a supervising broker hands + // to the proxy (host for dialing, path prefix for forwarding). For + // address-based entries every field stays zero/empty. + OpenAIUp bool `json:"openai_up"` + OpenAIBaseURL string `json:"openai_base_url,omitempty"` + OpenAIHost string `json:"openai_host,omitempty"` + OpenAIPort int `json:"openai_port,omitempty"` + OpenAIBasePath string `json:"openai_base_path,omitempty"` + OpenAIModels []string `json:"openai_models,omitempty"` + NodeInfoUp bool `json:"node_info_up"` NodeInfoPort int `json:"node_info_port"` TLSEnabled bool `json:"tls_enabled,omitempty"` MTLSRequired bool `json:"mtls_required,omitempty"` @@ -208,6 +227,10 @@ func (m *Manager) Run(ctx context.Context) error { m.cancel = cancel defer cancel() + // Restore the durable list before announcing ready, so supervisors that + // replay only their own state see the full node set immediately. + m.loadPersistedEntries() + if err := m.codec.Notify("ready", ReadyParams{Version: Version}); err != nil { return fmt.Errorf("failed to send ready notification: %w", err) } @@ -251,8 +274,30 @@ func (m *Manager) probeNode(entry ManualEntry) { addr := entry.Address id := nodeID(entry) - ollamaUp, ollamaModels := m.probeOllama(addr, 11434) - lmStudioUp, lmStudioModels := m.probeLMStudio(addr, lmStudioPort) + // A declared OpenAI endpoint is probed at its own URL (and node-info on + // the URL's host); the default engine-port legs only apply to address + // entries, which name a host that may run any of our engines. + ollamaUp, ollamaModels := false, []string(nil) + lmStudioUp, lmStudioModels := false, []string(nil) + openAIUp, openAIModels := false, []string(nil) + var openAI endpointTarget + if entry.OpenAIBaseURL != "" { + target, err := parseEndpointTarget(entry.OpenAIBaseURL) + if err != nil { + // Unreachable in practice — node/add validates the URL before + // the entry exists. Surface it as a down endpoint rather than + // crashing the probe loop. + slog.Warn("manual endpoint entry has an invalid base URL", + "node_id", id, "err", err) + } else { + openAI = target + addr = target.host + openAIUp, openAIModels = m.probeOpenAI(target) + } + } else { + ollamaUp, ollamaModels = m.probeOllama(addr, 11434) + lmStudioUp, lmStudioModels = m.probeLMStudio(addr, lmStudioPort) + } // Pick scheme + port + client based on the entry's TLS hint. // The operator decides which scheme this manual node uses; we @@ -290,6 +335,12 @@ func (m *Manager) probeNode(entry ManualEntry) { LMStudioUp: lmStudioUp, LMStudioPort: lmStudioPort, LMStudioModels: lmStudioModels, + OpenAIUp: openAIUp, + OpenAIBaseURL: entry.OpenAIBaseURL, + OpenAIHost: openAI.host, + OpenAIPort: openAI.port, + OpenAIBasePath: openAI.basePath, + OpenAIModels: openAIModels, NodeInfoUp: nodeInfoUp, NodeInfoPort: nodeInfoPort, TLSEnabled: entry.TLSPort > 0, @@ -302,7 +353,7 @@ func (m *Manager) probeNode(entry ManualEntry) { HostUUID: info.HostUUID, } - reachable := newStatus.OllamaUp || newStatus.LMStudioUp || newStatus.NodeInfoUp + reachable := newStatus.OllamaUp || newStatus.LMStudioUp || newStatus.OpenAIUp || newStatus.NodeInfoUp m.mu.Lock() tn, exists := m.nodes[id] @@ -331,10 +382,12 @@ func (m *Manager) probeNode(entry ManualEntry) { changed := prev.OllamaUp != newStatus.OllamaUp || prev.LMStudioUp != newStatus.LMStudioUp || + prev.OpenAIUp != newStatus.OpenAIUp || prev.NodeInfoUp != newStatus.NodeInfoUp || prev.HostUUID != newStatus.HostUUID || !sliceEqual(prev.OllamaModels, newStatus.OllamaModels) || !sliceEqual(prev.LMStudioModels, newStatus.LMStudioModels) || + !sliceEqual(prev.OpenAIModels, newStatus.OpenAIModels) || !gpusEqual(prev.GPUs, newStatus.GPUs) || !cpuEqual(prev.CPU, newStatus.CPU) || !memoryEqual(prev.Memory, newStatus.Memory) || @@ -403,23 +456,79 @@ func probeFailedID(nodeID string) string { // manager (which only governs the local engine). const lmStudioPort = 1234 +// endpointTarget is a parsed, validated openai_base_url. baseURL is the full +// OpenAI-client base (e.g. http://host:8888/v1); host/port serve the node-info +// probe and display; basePath is the URL's path prefix ("" or "/v1"), trailing +// slash stripped, that the proxy prepends to forwarded OpenAI paths. +type endpointTarget struct { + baseURL *url.URL + host string + port int + basePath string +} + +// parseEndpointTarget validates an operator-supplied OpenAI-compatible base +// URL. v1 accepts http only: the proxy dials manual nodes over plain HTTP. +func parseEndpointTarget(raw string) (endpointTarget, error) { + u, err := url.Parse(raw) + if err != nil { + return endpointTarget{}, fmt.Errorf("invalid URL %q: %w", raw, err) + } + if u.Scheme != "http" { + return endpointTarget{}, fmt.Errorf("unsupported scheme %q in %q (http only for now)", u.Scheme, raw) + } + if u.Hostname() == "" { + return endpointTarget{}, fmt.Errorf("missing host in %q", raw) + } + if u.RawQuery != "" || u.Fragment != "" { + return endpointTarget{}, fmt.Errorf("query or fragment not allowed in base URL %q", raw) + } + port := 80 + if p := u.Port(); p != "" { + port, err = strconv.Atoi(p) + if err != nil { + return endpointTarget{}, fmt.Errorf("invalid port in %q", raw) + } + } + return endpointTarget{ + baseURL: u, + host: u.Hostname(), + port: port, + basePath: strings.TrimSuffix(u.Path, "/"), + }, nil +} + // probeLMStudio checks LM Studio's OpenAI-compatible server on addr:port. A // single GET /v1/models doubles as the liveness check and the model list (the // response is {"data":[{"id":"..."}],...}). Returns whether it is up and the // model ids it serves. func (m *Manager) probeLMStudio(addr string, port int) (bool, []string) { - url := "http://" + net.JoinHostPort(addr, strconv.Itoa(port)) + "/v1/models" + return m.probeModelsAPI("http://"+net.JoinHostPort(addr, strconv.Itoa(port))+"/v1/models") +} + +// probeOpenAI GETs the declared endpoint's model list ({base}/models). Like +// probeLMStudio, one call doubles as liveness and inventory. +func (m *Manager) probeOpenAI(t endpointTarget) (bool, []string) { + return m.probeModelsAPI(t.baseURL.String() + "/models") +} + +// probeModelsAPI checks an OpenAI-compatible model list endpoint. A single GET +// doubles as the liveness check and the model list (the response is +// {"data":[{"id":"..."}],...}). Returns whether it is up and the model ids it +// serves; a reachable endpoint whose list doesn't parse reports up with no +// models. +func (m *Manager) probeModelsAPI(modelsURL string) (bool, []string) { start := time.Now() - resp, err := m.client.Get(url) + resp, err := m.client.Get(modelsURL) if err != nil { - slog.Debug("manual probe lmstudio failed", - "addr", addr, "port", port, "duration_ms", time.Since(start).Milliseconds(), "err", err) + slog.Debug("manual probe models failed", + "url", modelsURL, "duration_ms", time.Since(start).Milliseconds(), "err", err) return false, nil } defer resp.Body.Close() if resp.StatusCode != http.StatusOK { - slog.Debug("manual probe lmstudio non-OK", - "addr", addr, "port", port, "status", resp.StatusCode, + slog.Debug("manual probe models non-OK", + "url", modelsURL, "status", resp.StatusCode, "duration_ms", time.Since(start).Milliseconds()) return false, nil } @@ -430,8 +539,8 @@ func (m *Manager) probeLMStudio(addr string, port int) (bool, []string) { } if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { // Reachable, but the model list didn't parse — still report it up. - slog.Debug("manual probe lmstudio up (models parse failed)", - "addr", addr, "port", port, "err", err) + slog.Debug("manual probe models up (models parse failed)", + "url", modelsURL, "err", err) return true, nil } models := make([]string, 0, len(result.Data)) @@ -440,8 +549,8 @@ func (m *Manager) probeLMStudio(addr string, port int) (bool, []string) { models = append(models, d.ID) } } - slog.Debug("manual probe lmstudio up", - "addr", addr, "port", port, "models", len(models), + slog.Debug("manual probe models up", + "url", modelsURL, "models", len(models), "duration_ms", time.Since(start).Milliseconds()) return true, models } @@ -544,6 +653,18 @@ func (m *Manager) addNode(entry ManualEntry) ManualNodeStatus { TLSEnabled: entry.TLSPort > 0, MTLSRequired: entry.TLSPort > 0 && entry.MTLS, } + // Pre-echo the endpoint's configured parts (and the URL's host as the + // display address) so the unprobed initial status already carries the + // operator's configuration, mirroring the fixed-port pre-echoes above. + if entry.OpenAIBaseURL != "" { + if t, err := parseEndpointTarget(entry.OpenAIBaseURL); err == nil { + status.Address = t.host + status.OpenAIBaseURL = entry.OpenAIBaseURL + status.OpenAIHost = t.host + status.OpenAIPort = t.port + status.OpenAIBasePath = t.basePath + } + } m.mu.Lock() m.nodes[id] = &trackedNode{entry: entry, status: status} @@ -645,17 +766,26 @@ func (m *Manager) handleMessage(msg *Message) { case "node/add": var entry ManualEntry if err := json.Unmarshal(msg.Params, &entry); err != nil { - m.codec.RespondError(msg.ID, -32602, "invalid params: expected {\"address\": \"...\"}") + m.codec.RespondError(msg.ID, -32602, "invalid params: expected {\"address\": \"...\"} or {\"openai_base_url\": \"...\"}") return } - if entry.Address == "" { - m.codec.RespondError(msg.ID, -32602, "address is required") + hasAddress := entry.Address != "" + hasBaseURL := entry.OpenAIBaseURL != "" + if hasAddress == hasBaseURL { + m.codec.RespondError(msg.ID, -32602, "exactly one of address or openai_base_url is required") return } + if hasBaseURL { + if _, err := parseEndpointTarget(entry.OpenAIBaseURL); err != nil { + m.codec.RespondError(msg.ID, -32602, err.Error()) + return + } + } status := m.addNode(entry) if err := m.codec.Respond(msg.ID, status); err != nil { log.Printf("failed to respond to node/add: %v", err) } + m.saveNow() log.Printf("manual node added: %s (%s)", status.ID, entry.Address) case "node/remove": @@ -671,6 +801,7 @@ func (m *Manager) handleMessage(msg *Message) { log.Printf("failed to respond to node/remove: %v", err) } if removed { + m.saveNow() log.Printf("manual node removed: %s", params.ID) } @@ -698,6 +829,12 @@ func nodeID(entry ManualEntry) string { if entry.Name != "" { return entry.Name } + if entry.OpenAIBaseURL != "" { + if t, err := parseEndpointTarget(entry.OpenAIBaseURL); err == nil { + return "manual:" + net.JoinHostPort(t.host, strconv.Itoa(t.port)) + } + return "manual:" + entry.OpenAIBaseURL + } return "manual:" + entry.Address } diff --git a/services/nvpair-manual-nodes/manager_test.go b/services/nvpair-manual-nodes/manager_test.go index e52d9d5d..543ec3d5 100644 --- a/services/nvpair-manual-nodes/manager_test.go +++ b/services/nvpair-manual-nodes/manager_test.go @@ -689,3 +689,156 @@ func writePipeRequest(t *testing.T, conn net.Conn, id int, method string, params t.Fatalf("write request: %v", err) } } + +func TestParseEndpointTarget(t *testing.T) { + cases := []struct { + raw, host string + port int + basePath string + wantErr bool + }{ + {raw: "http://localhost:8888/v1", host: "localhost", port: 8888, basePath: "/v1"}, + {raw: "http://dgx:8000", host: "dgx", port: 8000, basePath: ""}, + {raw: "http://dgx", host: "dgx", port: 80, basePath: ""}, + {raw: "http://dgx:8000/", host: "dgx", port: 8000, basePath: ""}, + {raw: "https://localhost:8888/v1", wantErr: true}, + {raw: "http://:8888/v1", wantErr: true}, + {raw: "http://localhost:8888/v1?x=1", wantErr: true}, + {raw: "not a url", wantErr: true}, + } + for _, tc := range cases { + got, err := parseEndpointTarget(tc.raw) + if tc.wantErr { + if err == nil { + t.Fatalf("parseEndpointTarget(%q): expected error, got %+v", tc.raw, got) + } + continue + } + if err != nil { + t.Fatalf("parseEndpointTarget(%q): %v", tc.raw, err) + } + if got.host != tc.host || got.port != tc.port || got.basePath != tc.basePath { + t.Fatalf("parseEndpointTarget(%q) = %+v, want host=%s port=%d basePath=%q", + tc.raw, got, tc.host, tc.port, tc.basePath) + } + } +} + +func TestProbeOpenAI(t *testing.T) { + m, _, rt := newTestManager() + rt.set("GET", "localhost:8888", "/v1/models", func(*http.Request) (*http.Response, error) { + return httpJSON(http.StatusOK, `{"object":"list","data":[{"id":"m1"},{"id":"m2"}]}`) + }) + up, models := m.probeOpenAI(mustParseForTest(t, "http://localhost:8888/v1")) + if !up || len(models) != 2 || models[0] != "m1" || models[1] != "m2" { + t.Fatalf("probeOpenAI = %v %v, want up [m1 m2]", up, models) + } + + rt.set("GET", "other:9999", "/v1/models", func(*http.Request) (*http.Response, error) { + return httpJSON(http.StatusNotFound, `{"error":"nope"}`) + }) + up, models = m.probeOpenAI(mustParseForTest(t, "http://other:9999/v1")) + if up || models != nil { + t.Fatalf("probeOpenAI down case = %v %v, want false nil", up, models) + } + + // Reachable but unparseable: up with no models (same contract as probeLMStudio). + rt.set("GET", "weird:7777", "/v1/models", func(*http.Request) (*http.Response, error) { + return httpJSON(http.StatusOK, `not json`) + }) + up, models = m.probeOpenAI(mustParseForTest(t, "http://weird:7777/v1")) + if !up || models != nil { + t.Fatalf("probeOpenAI unparseable case = %v %v, want up nil", up, models) + } +} + +// mustParseForTest is a test-only helper: parseEndpointTarget or t.Fatal. +func mustParseForTest(t *testing.T, raw string) endpointTarget { + t.Helper() + got, err := parseEndpointTarget(raw) + if err != nil { + t.Fatalf("parseEndpointTarget(%q): %v", raw, err) + } + return got +} + +func TestNodeIDForEndpointEntry(t *testing.T) { + if got := nodeID(ManualEntry{OpenAIBaseURL: "http://vllm-box:8888/v1"}); got != "manual:vllm-box:8888" { + t.Fatalf("endpoint nodeID = %q", got) + } + if got := nodeID(ManualEntry{Name: "my-vllm", OpenAIBaseURL: "http://vllm-box:8888/v1"}); got != "my-vllm" { + t.Fatalf("named endpoint nodeID = %q", got) + } +} + +func TestAddEndpointNode(t *testing.T) { + m, rw, rt := newTestManager() + rt.set("GET", "localhost:8888", "/v1/models", func(*http.Request) (*http.Response, error) { + return httpJSON(http.StatusOK, `{"object":"list","data":[{"id":"stub-model"}]}`) + }) + rt.set("GET", "localhost:14318", "/v1/node-info", func(*http.Request) (*http.Response, error) { + data, _ := json.Marshal(sampleInfo()) + return httpJSON(http.StatusOK, string(data)) + }) + + m.handleMessage(requestMessage(1, "node/add", ManualEntry{OpenAIBaseURL: "http://localhost:8888/v1", Name: "my-vllm"})) + + resp := readCaptureUntil(t, rw, responseWithID(1)) + initial := decodeResult[ManualNodeStatus](t, resp) + if initial.ID != "my-vllm" { + t.Fatalf("initial status id = %+v", initial) + } + if initial.Address != "localhost" { + t.Fatalf("initial status address = %q, want localhost (the URL's host)", initial.Address) + } + if initial.OpenAIBaseURL != "http://localhost:8888/v1" || initial.OpenAIHost != "localhost" || + initial.OpenAIPort != 8888 || initial.OpenAIBasePath != "/v1" { + t.Fatalf("initial status endpoint echo = %+v", initial) + } + if initial.OpenAIUp { + t.Fatalf("initial status should be unprobed: %+v", initial) + } + + discovered := readCaptureUntil(t, rw, methodIs("node/discovered")) + status := decodeParams[ManualNodeStatus](t, discovered) + if !status.OpenAIUp || !status.NodeInfoUp { + t.Fatalf("discovered status did not include openai endpoint + node-info: %+v", status) + } + if len(status.OpenAIModels) != 1 || status.OpenAIModels[0] != "stub-model" { + t.Fatalf("openai models = %#v", status.OpenAIModels) + } + if status.OpenAIBasePath != "/v1" || status.OpenAIHost != "localhost" || status.OpenAIPort != 8888 { + t.Fatalf("discovered endpoint fields = %+v", status) + } + if status.OllamaUp || status.LMStudioUp { + t.Fatalf("endpoint entry must not probe the default engine ports: %+v", status) + } + if !status.TelemetryValid || status.MSSince != 137 { + t.Fatalf("telemetry = valid:%v age:%d, want true/137", status.TelemetryValid, status.MSSince) + } +} + +func TestAddNodeRequiresExactlyOneOfAddressOrBaseURL(t *testing.T) { + m, rw, _ := newTestManager() + + m.handleMessage(requestMessage(1, "node/add", + ManualEntry{Address: "node.local", OpenAIBaseURL: "http://node.local:8888/v1"})) + resp := readCaptureFrame(t, rw) + if resp.Error == nil || resp.Error.Code != -32602 { + t.Fatalf("both address and base URL error = %+v", resp.Error) + } + + m.handleMessage(requestMessage(2, "node/add", + ManualEntry{OpenAIBaseURL: "https://node.local:8888/v1"})) + resp = readCaptureFrame(t, rw) + if resp.Error == nil || resp.Error.Code != -32602 { + t.Fatalf("https base URL error = %+v", resp.Error) + } + if resp.Error != nil && !strings.Contains(resp.Error.Message, "http only") { + t.Fatalf("https error message %q should say http only", resp.Error.Message) + } + + if got := m.listNodes(); len(got) != 0 { + t.Fatalf("validation errors added nodes: %#v", got) + } +} diff --git a/services/nvpair-manual-nodes/store.go b/services/nvpair-manual-nodes/store.go new file mode 100644 index 00000000..88c390a2 --- /dev/null +++ b/services/nvpair-manual-nodes/store.go @@ -0,0 +1,110 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +package main + +import ( + "encoding/json" + "log/slog" + "os" + "path/filepath" + "sort" + + "nvpair-shared/appdir" +) + +// entriesFile is the on-disk file name of the service-owned durable +// manual-node list, inside the shared app data dir (appdir). +const entriesFile = "manual-nodes.json" + +// entriesFileShape is the on-disk shape of the durable list. +type entriesFileShape struct { + Entries []ManualEntry `json:"entries"` +} + +func entriesPath() (string, error) { return appdir.Path(entriesFile) } + +// loadEntries returns the persisted entries. A missing file reports nil, nil +// (first run); a corrupt file reports an error the caller logs and ignores +// (the in-memory list stays authoritative for this run). +func loadEntries() ([]ManualEntry, error) { + path, err := entriesPath() + if err != nil { + return nil, err + } + data, err := os.ReadFile(path) + if os.IsNotExist(err) { + return nil, nil + } + if err != nil { + return nil, err + } + var f entriesFileShape + if err := json.Unmarshal(data, &f); err != nil { + return nil, err + } + return f.Entries, nil +} + +// saveEntries atomically writes the full entry set (tmp + rename) so a crash +// mid-write can't leave a truncated file behind. +func saveEntries(entries []ManualEntry) error { + path, err := entriesPath() + if err != nil { + return err + } + if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { + return err + } + data, err := json.Marshal(entriesFileShape{Entries: entries}) + if err != nil { + return err + } + tmp := path + ".tmp" + if err := os.WriteFile(tmp, data, 0o644); err != nil { + return err + } + if err := os.Rename(tmp, path); err != nil { + _ = os.Remove(tmp) + return err + } + return nil +} + +// currentEntries returns the tracked entries sorted by node id so the +// persisted file is deterministic across saves. +func (m *Manager) currentEntries() []ManualEntry { + m.mu.RLock() + defer m.mu.RUnlock() + out := make([]ManualEntry, 0, len(m.nodes)) + for _, tn := range m.nodes { + out = append(out, tn.entry) + } + sort.Slice(out, func(i, j int) bool { return nodeID(out[i]) < nodeID(out[j]) }) + return out +} + +// loadPersistedEntries restores the durable list into memory at startup, so +// manual nodes added through any frontend survive a restart of this service. +// Each restored entry is probed and announced exactly like a fresh node/add. +func (m *Manager) loadPersistedEntries() { + entries, err := loadEntries() + if err != nil { + slog.Warn("failed to load persisted manual entries", "err", err) + return + } + for _, e := range entries { + m.addNode(e) + } + if len(entries) > 0 { + slog.Info("manual entries restored", "count", len(entries)) + } +} + +// saveNow persists the current entry set; a failed save means an entry won't +// survive the next restart, which the operator can re-add — log, don't fail. +func (m *Manager) saveNow() { + if err := saveEntries(m.currentEntries()); err != nil { + slog.Warn("failed to save manual entries", "err", err) + } +} diff --git a/services/nvpair-manual-nodes/store_test.go b/services/nvpair-manual-nodes/store_test.go new file mode 100644 index 00000000..ec9db217 --- /dev/null +++ b/services/nvpair-manual-nodes/store_test.go @@ -0,0 +1,112 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +package main + +import ( + "os" + "path/filepath" + "testing" + + "nvpair-shared/appdir" +) + +// withIsolatedAppdir points the shared appdir (and thus the entries file) at a +// fresh temp dir for the duration of the test. +func withIsolatedAppdir(t *testing.T) { + t.Helper() + t.Setenv("XDG_CONFIG_HOME", t.TempDir()) +} + +func entriesFilePath(t *testing.T) string { + t.Helper() + path, err := appdir.Path(entriesFile) + if err != nil { + t.Fatalf("appdir path: %v", err) + } + return path +} + +func TestEntriesRoundTrip(t *testing.T) { + withIsolatedAppdir(t) + entries := []ManualEntry{ + {Address: "h1", Name: "one"}, + {OpenAIBaseURL: "http://h2:8888/v1", Name: "two"}, + } + if err := saveEntries(entries); err != nil { + t.Fatalf("saveEntries: %v", err) + } + got, err := loadEntries() + if err != nil { + t.Fatalf("loadEntries: %v", err) + } + if len(got) != 2 || got[0].Address != "h1" || got[1].OpenAIBaseURL != "http://h2:8888/v1" { + t.Fatalf("round trip = %+v", got) + } +} + +func TestSaveEntriesIsAtomicNoTmpLeftBehind(t *testing.T) { + withIsolatedAppdir(t) + if err := saveEntries([]ManualEntry{{Address: "h1"}}); err != nil { + t.Fatalf("saveEntries: %v", err) + } + dir := filepath.Dir(entriesFilePath(t)) + entries, err := os.ReadDir(dir) + if err != nil { + t.Fatalf("read dir: %v", err) + } + for _, e := range entries { + if filepath.Ext(e.Name()) == ".tmp" { + t.Fatalf("leftover tmp file: %s", e.Name()) + } + } +} + +func TestLoadEntriesMissingFile(t *testing.T) { + withIsolatedAppdir(t) + entries, err := loadEntries() + if err != nil || entries != nil { + t.Fatalf("missing file = %v, %v; want nil, nil", entries, err) + } +} + +func TestLoadEntriesCorrupt(t *testing.T) { + withIsolatedAppdir(t) + path := entriesFilePath(t) + if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(path, []byte("not json"), 0o644); err != nil { + t.Fatal(err) + } + if _, err := loadEntries(); err == nil { + t.Fatal("corrupt file: expected error") + } +} + +// TestManagerPersistsAcrossRestart covers the durable-list contract: an entry +// added through node/add survives a "restart" (a fresh manager loading from +// disk), and a node/remove is persisted too. +func TestManagerPersistsAcrossRestart(t *testing.T) { + withIsolatedAppdir(t) + + m1, rw1, _ := newTestManager() + m1.handleMessage(requestMessage(1, "node/add", ManualEntry{Address: "h1", Name: "one"})) + _ = readCaptureUntil(t, rw1, responseWithID(1)) + + m2, rw2, _ := newTestManager() + m2.loadPersistedEntries() + if got := m2.listNodes(); len(got) != 1 || got[0].ID != "one" { + t.Fatalf("restored entries = %+v, want one", got) + } + + m2.handleMessage(requestMessage(2, "node/remove", map[string]string{"id": "one"})) + _ = readCaptureUntil(t, rw2, responseWithID(2)) + got, err := loadEntries() + if err != nil { + t.Fatalf("loadEntries after remove: %v", err) + } + if len(got) != 0 { + t.Fatalf("entry still persisted after remove: %+v", got) + } +} From 4e688e743b36e4d28a46f6889b00c92d217b4371 Mon Sep 17 00:00:00 2001 From: Will Ford Date: Mon, 7 Sep 2026 23:11:54 +0200 Subject: [PATCH 2/7] ui-broker: bridge OpenAI endpoints into lmstudio-proxy with base path Declared endpoints flow through to the proxy carrying their path prefix; their model lists attribute to the new 'openai' engine in modelsByEngine, and node/remove by UUID translates back to the manual store key for the endpoint leg. Endpoint entries keep their manual identity (the declared URL) and never adopt the host's learned node-info UUID, so endpoints on the same box as PAIR stay separate nodes instead of clobbering each other. Signed-off-by: Will Ford --- services/nvpair-ui-broker/broker.go | 62 ++++++- services/nvpair-ui-broker/manualnodes.go | 68 ++++++-- services/nvpair-ui-broker/manualnodes_test.go | 164 ++++++++++++++++++ services/tests/broker_management_test.go | 79 +++++++++ 4 files changed, 357 insertions(+), 16 deletions(-) create mode 100644 services/nvpair-ui-broker/manualnodes_test.go diff --git a/services/nvpair-ui-broker/broker.go b/services/nvpair-ui-broker/broker.go index 0d189578..8f2eb8d2 100644 --- a/services/nvpair-ui-broker/broker.go +++ b/services/nvpair-ui-broker/broker.go @@ -1256,11 +1256,50 @@ func (b *Broker) forwardManualNodesNotification(method string, params json.RawMe // node-info first reported its real hostUuid, moving it off the manual id). On a // rekey it drops the manual claim under the old key (the scanner's claim, if // any, survives); the caller rebridges the proxy overlay off the old key. +// manualAliasForStoreKey inverts the alias->storeKey tracking: given a +// discovery-store key (typically the node's hostUuid), return the manual alias +// id the manual-nodes service tracks it under. The desktop removes nodes by +// UUID, but the service keys its entries by their add-time id, so the relay +// translates before forwarding node/remove. When several aliases share a key +// (two entries for one machine), the lexicographically smallest is chosen — +// the same deterministic rule survivingAliasLocked uses. +func (b *Broker) manualAliasForStoreKey(storeKey string) (string, bool) { + b.manualMu.Lock() + defer b.manualMu.Unlock() + best := "" + for alias, key := range b.manualNodeKeys { + if key == storeKey && (best == "" || alias < best) { + best = alias + } + } + if best == "" { + return "", false + } + return best, true +} + +// translateManualRemoveID rebinds a node/remove's id from a discovery-store key +// to the manual alias id the service tracks, when lookup resolves it to a +// different alias. Returns the rewritten params, or (nil, false) when the id +// passes through untouched (not a known store key, or already the alias id). +func translateManualRemoveID(id string, lookup func(storeKey string) (string, bool)) (json.RawMessage, bool) { + alias, ok := lookup(id) + if !ok || alias == id { + return nil, false + } + params, err := json.Marshal(map[string]string{"id": alias}) + if err != nil { + return nil, false + } + return params, true +} + // upsertManualNode ingests a manual node status: it records the alias's payload, // keys the discovery store by the node's operational identity (its hostUuid once -// node-info reports one, else the manual id), and bridges the proxy candidate -// under that same key so scheduler priority and scheduledOn resolve to it. If -// the alias's key changed (node-info revealed its real UUID) the +// node-info reports one, else the manual id — declared endpoints always keep +// their manual id, see manualToEnriched), and bridges the proxy candidate under +// that same key so scheduler priority and scheduledOn resolve to it. If the +// alias's key changed (node-info revealed its real UUID) the // old key is reprojected from a surviving alias or released. func (b *Broker) upsertManualNode(s manualNodeStatus) { en := manualToEnriched(s) @@ -3159,7 +3198,22 @@ func (b *Broker) relayToManualNodes(msg *Message) { } id := msg.ID method := msg.Method - relayErr := mn.RelayRequest(method, msg.Params, func(result json.RawMessage, rpcErr *RPCError, err error) { + // The desktop removes nodes by UUID (the discovery-store key). A manual + // alias rekeyed to its node-info hostUuid is still tracked by the service + // under its add-time id, so translate store key -> alias id first; ids the + // broker doesn't recognize pass through untouched. + relayParams := msg.Params + if method == "node/remove" { + var p struct { + ID string `json:"id"` + } + if json.Unmarshal(msg.Params, &p) == nil && p.ID != "" { + if rewritten, ok := translateManualRemoveID(p.ID, b.manualAliasForStoreKey); ok { + relayParams = rewritten + } + } + } + relayErr := mn.RelayRequest(method, relayParams, func(result json.RawMessage, rpcErr *RPCError, err error) { switch { case err != nil: if e := b.codec.RespondError(id, -32000, fmt.Sprintf("manual-nodes call failed: %v", err)); e != nil { diff --git a/services/nvpair-ui-broker/manualnodes.go b/services/nvpair-ui-broker/manualnodes.go index 47fe0ff1..4b35f2ec 100644 --- a/services/nvpair-ui-broker/manualnodes.go +++ b/services/nvpair-ui-broker/manualnodes.go @@ -28,7 +28,18 @@ type manualNodeStatus struct { LMStudioUp bool `json:"lmstudio_up"` LMStudioPort int `json:"lmstudio_port"` LMStudioModels []string `json:"lmstudio_models,omitempty"` - NodeInfoPort int `json:"node_info_port"` + // The openai_* fields mirror a declared OpenAI-compatible endpoint + // (an entry added with openai_base_url). OpenAIBaseURL is the operator's + // declaration (echoed for display); OpenAIHost/Port and OpenAIBasePath + // are the parsed parts the bridge hands to the proxy. Tags match the + // prober's producer so the payloads unmarshal straight across. + OpenAIUp bool `json:"openai_up"` + OpenAIBaseURL string `json:"openai_base_url,omitempty"` + OpenAIHost string `json:"openai_host,omitempty"` + OpenAIPort int `json:"openai_port,omitempty"` + OpenAIBasePath string `json:"openai_base_path,omitempty"` + OpenAIModels []string `json:"openai_models,omitempty"` + NodeInfoPort int `json:"node_info_port"` GPUs []GPUInfo `json:"gpus"` CPU *CPUInfo `json:"cpu"` Memory *MemoryInfo `json:"memory"` @@ -75,9 +86,17 @@ func manualToEnriched(s manualNodeStatus) EnrichedNode { // manual-node ingestion boundary — downstream keys off HostUUID with no // fallback. Once the real UUID is learned, a manually-added machine // that's also discovered over mDNS collapses to the one hostUuid-keyed entry. - hostUUID := s.HostUUID - if hostUUID == "" { - hostUUID = s.ID + // + // Declared endpoints are the exception: an endpoint's identity is the URL + // the operator declared, not the PAIR identity of the machine that happens + // to run it. An endpoint on this host (or on a peer that also runs + // node-info) must keep its manual key or it folds into that machine's own + // node — and two endpoints on one host collapse into a single slot that + // clobbers itself on every probe. Node-info hardware data still merges; + // only the key stays manual. + hostUUID := s.ID + if s.OpenAIBaseURL == "" && s.HostUUID != "" { + hostUUID = s.HostUUID } en := EnrichedNode{ ID: s.ID, @@ -87,7 +106,7 @@ func manualToEnriched(s manualNodeStatus) EnrichedNode { GPUs: s.GPUs, CPU: s.CPU, Memory: s.Memory, - Models: mergeModels(s.OllamaModels, s.LMStudioModels), + Models: mergeModels(s.OllamaModels, s.LMStudioModels, s.OpenAIModels), ModelsByEngine: manualModelsByEngine(s), } if s.Address != "" { @@ -99,8 +118,10 @@ func manualToEnriched(s manualNodeStatus) EnrichedNode { // manualModelsByEngine builds the per-engine attribution for a manual node from // the per-engine lists the prober already collected, keyed by the same // engine-manager engine names discovered nodes use ("ollama", "lmstudio") so the -// two discovery sources present ModelsByEngine identically. An engine with no -// models adds no key; returns nil when neither engine reports any. +// two discovery sources present ModelsByEngine identically. A declared +// OpenAI-compatible endpoint attributes under "openai" — the node isn't LM +// Studio, and the label must say what actually serves it. An engine with no +// models adds no key; returns nil when no engine reports any. func manualModelsByEngine(s manualNodeStatus) map[string][]string { byEngine := map[string][]string{} if len(s.OllamaModels) > 0 { @@ -109,6 +130,9 @@ func manualModelsByEngine(s manualNodeStatus) map[string][]string { if len(s.LMStudioModels) > 0 { byEngine["lmstudio"] = s.LMStudioModels } + if len(s.OpenAIModels) > 0 { + byEngine["openai"] = s.OpenAIModels + } if len(byEngine) == 0 { return nil } @@ -145,6 +169,10 @@ type proxyManualNode struct { Addresses []string `json:"addresses"` TXT []string `json:"txt,omitempty"` Models []string `json:"models,omitempty"` + // BasePath is the declared endpoint's API prefix (e.g. "/v1"); the proxy + // joins it onto its own /v1 root when forwarding. Empty for classic + // manual nodes, whose engines serve the OpenAI API at their own /v1 root. + BasePath string `json:"base_path,omitempty"` } // bridgeManualNode keeps every supervised proxy's manual-node set in step with @@ -160,17 +188,28 @@ type proxyManualNode struct { // daemon to carry — so without this explicit add the proxies can't route // inference to them even though both workers are broker-owned. func (b *Broker) bridgeManualNode(s manualNodeStatus, key string) { - b.bridgeToProxy(b.getProxy(), "ollama", s, key, s.OllamaUp, s.OllamaPort, s.OllamaModels) - b.bridgeToProxy(b.getLMStudioProxy(), "lmstudio", s, key, s.LMStudioUp, s.LMStudioPort, s.LMStudioModels) + b.bridgeToProxy(b.getProxy(), "ollama", s, key, s.OllamaUp, s.OllamaPort, s.OllamaModels, "") + b.bridgeToProxy(b.getLMStudioProxy(), "lmstudio", s, key, s.LMStudioUp, s.LMStudioPort, s.LMStudioModels, "") + // A declared OpenAI endpoint rides the same OpenAI proxy as LM Studio, + // but carries its base path so the proxy can join it onto forwarded + // paths, and it is labeled "openai", not "lmstudio". The leg is gated on + // a declared base URL: a classic address entry must not issue an openai + // remove — the remove would land on the same proxy+key the lmstudio leg + // just added and evict the node it was meant to keep. + if s.OpenAIBaseURL != "" { + b.bridgeToProxy(b.getLMStudioProxy(), "openai", s, key, s.OpenAIUp, s.OpenAIPort, s.OpenAIModels, s.OpenAIBasePath) + } } // bridgeToProxy adds the node to p when its engine is reachable, or removes it // otherwise. up/port/models are the engine-specific fields the caller pulled -// off the node's status. The proxy candidate is keyed by `key` — the node's +// off the node's status; basePath is the endpoint's API prefix ("" for +// classic manual nodes). The proxy candidate is keyed by `key` — the node's // operational identity (its hostUuid once node-info reports it, else the manual -// id) — the same key the discovery store and scheduler use, so the scheduler's +// id; declared endpoints keep their manual id, see manualToEnriched) — the +// same key the discovery store and scheduler use, so the scheduler's // priority list and scheduledOn resolve to this candidate. -func (b *Broker) bridgeToProxy(p *proxyProcess, engine string, s manualNodeStatus, key string, up bool, port int, models []string) { +func (b *Broker) bridgeToProxy(p *proxyProcess, engine string, s manualNodeStatus, key string, up bool, port int, models []string, basePath string) { if p == nil { return } @@ -181,6 +220,7 @@ func (b *Broker) bridgeToProxy(p *proxyProcess, engine string, s manualNodeStatu Port: port, Addresses: []string{s.Address}, Models: models, + BasePath: basePath, } b.callProxyManual(p, engine, "node/add-manual", node, key) return @@ -196,6 +236,10 @@ func (b *Broker) bridgeToProxy(p *proxyProcess, engine string, s manualNodeStatu func (b *Broker) removeManualNodeFromProxies(id string) { b.callProxyManual(b.getProxy(), "ollama", "node/remove-manual", map[string]string{"id": id}, id) b.callProxyManual(b.getLMStudioProxy(), "lmstudio", "node/remove-manual", map[string]string{"id": id}, id) + // The openai leg shares lmstudio-proxy with the leg above; the duplicate + // remove is an idempotent no-op there, and keeps a standalone endpoint + // entry (never bridged as lmstudio) fully cleaned up. + b.callProxyManual(b.getLMStudioProxy(), "openai", "node/remove-manual", map[string]string{"id": id}, id) } // callProxyManual issues a best-effort node/add-manual|remove-manual to a diff --git a/services/nvpair-ui-broker/manualnodes_test.go b/services/nvpair-ui-broker/manualnodes_test.go new file mode 100644 index 00000000..6b898eeb --- /dev/null +++ b/services/nvpair-ui-broker/manualnodes_test.go @@ -0,0 +1,164 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +package main + +import ( + "encoding/json" + "testing" +) + +// TestManualToEnrichedIncludesOpenAI: an endpoint's models merge into the +// node's model list and attribute under the "openai" engine key (the node is +// not LM Studio), alongside the historical ollama/lmstudio attribution. +func TestManualToEnrichedIncludesOpenAI(t *testing.T) { + s := manualNodeStatus{ + ID: "stub", + Address: "10.0.1.9", + OpenAIUp: true, + OpenAIBaseURL: "http://10.0.1.9:8888/v1", + OpenAIHost: "10.0.1.9", + OpenAIPort: 8888, + OpenAIBasePath: "/v1", + OpenAIModels: []string{"m1", "m2"}, + NodeInfoPort: 14318, + } + en := manualToEnriched(s) + if len(en.Models) != 2 || en.Models[0] != "m1" || en.Models[1] != "m2" { + t.Fatalf("Models = %#v, want [m1 m2]", en.Models) + } + if got := en.ModelsByEngine["openai"]; len(got) != 2 || got[0] != "m1" || got[1] != "m2" { + t.Fatalf("ModelsByEngine[openai] = %#v, want [m1 m2]", got) + } + if _, hasOllama := en.ModelsByEngine["ollama"]; hasOllama { + t.Fatalf("no ollama key expected: %#v", en.ModelsByEngine) + } + + // An address-based node with only ollama models is unchanged: no "openai" key. + plain := manualNodeStatus{ID: "p", Address: "10.0.1.10", OllamaModels: []string{"o1"}, NodeInfoPort: 14318} + en = manualToEnriched(plain) + if _, has := en.ModelsByEngine["openai"]; has { + t.Fatalf("unexpected openai key for address entry: %#v", en.ModelsByEngine) + } + if got := en.ModelsByEngine["ollama"]; len(got) != 1 || got[0] != "o1" { + t.Fatalf("ModelsByEngine[ollama] = %#v, want [o1]", got) + } +} + +// TestManualModelsByEngineOpenAIOnly: an endpoint with no other engines +// reports exactly the "openai" attribution. +func TestManualModelsByEngineOpenAIOnly(t *testing.T) { + byEngine := manualModelsByEngine(manualNodeStatus{OpenAIModels: []string{"m1"}}) + if len(byEngine) != 1 { + t.Fatalf("byEngine = %#v, want exactly one key", byEngine) + } + if got := byEngine["openai"]; len(got) != 1 || got[0] != "m1" { + t.Fatalf("openai attribution = %#v, want [m1]", got) + } +} + +// TestManualAliasForStoreKey: the node/remove translation maps a store key +// (hostUuid) back to the alias id the manual-nodes service tracks, choosing +// the lexicographically smallest alias when several share a key. +func TestManualAliasForStoreKey(t *testing.T) { + b := newManualTestBroker() + b.upsertManualNode(manualStatus("alias-b", "10.0.0.2", "uuid-1")) + b.upsertManualNode(manualStatus("alias-a", "10.0.0.1", "uuid-1")) + b.upsertManualNode(manualStatus("alias-c", "10.0.0.3", "uuid-2")) + + if got, ok := b.manualAliasForStoreKey("uuid-1"); !ok || got != "alias-a" { + t.Fatalf("manualAliasForStoreKey(uuid-1) = %q, %v; want alias-a true", got, ok) + } + if got, ok := b.manualAliasForStoreKey("uuid-2"); !ok || got != "alias-c" { + t.Fatalf("manualAliasForStoreKey(uuid-2) = %q, %v; want alias-c true", got, ok) + } + if got, ok := b.manualAliasForStoreKey("unknown"); ok || got != "" { + t.Fatalf("manualAliasForStoreKey(unknown) = %q, %v; want empty false", got, ok) + } +} + +// TestRelayRemovesTranslateStoreKey: a node/remove arriving keyed by a store +// key (the desktop removes by UUID) is rewritten to the alias id before +// relaying; ids the broker doesn't recognize pass through untouched. +func TestRelayRemovesTranslateStoreKey(t *testing.T) { + cases := []struct { + name string + existing string // alias known to own store key "uuid-1" ("" = none) + in string + wantRewrite bool + want string + }{ + {name: "store-key-translated", existing: "alias-a", in: "uuid-1", wantRewrite: true, want: "alias-a"}, + {name: "alias-id-passthrough", existing: "alias-a", in: "alias-a"}, + {name: "unknown-id-passthrough", existing: "alias-a", in: "someone-else"}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + got, ok := translateManualRemoveID(tc.in, func(storeKey string) (string, bool) { + if tc.existing != "" && storeKey == "uuid-1" { + return tc.existing, true + } + return "", false + }) + if tc.wantRewrite { + if !ok { + t.Fatalf("translateManualRemoveID(%q): expected rewrite", tc.in) + } + var out struct { + ID string `json:"id"` + } + if err := json.Unmarshal(got, &out); err != nil { + t.Fatalf("unmarshal rewritten params %q: %v", got, err) + } + if out.ID != tc.want { + t.Fatalf("rewritten id = %q, want %q", out.ID, tc.want) + } + return + } + // Passthrough: no rewrite signal, so the relay keeps the original + // params (id unchanged). + if ok { + t.Fatalf("translateManualRemoveID(%q): unexpected rewrite to %s", tc.in, got) + } + }) + } +} + +// TestManualToEnrichedEndpointKeepsManualKey pins the endpoint identity rule: +// a declared endpoint's operational key is the manual id (the declared URL's +// identity), never the host's learned node-info UUID. Endpoints commonly sit on +// the same box as the PAIR installation (or on a peer that also runs +// node-info); adopting the host's UUID would fold the endpoint into that +// machine's own node, and two endpoints on one host would collapse into a +// single slot that clobbers itself on every probe cycle. +func TestManualToEnrichedEndpointKeepsManualKey(t *testing.T) { + s := manualNodeStatus{ + ID: "manual:127.0.0.1:8888", + Address: "127.0.0.1", + OpenAIUp: true, + OpenAIBaseURL: "http://127.0.0.1:8888/v1", + OpenAIHost: "127.0.0.1", + OpenAIPort: 8888, + OpenAIBasePath: "/v1", + OpenAIModels: []string{"m1"}, + NodeInfoPort: 14318, + HostUUID: "learned-host-uuid", + } + if got := manualToEnriched(s).storeKey(); got != s.ID { + t.Fatalf("endpoint storeKey = %q, want the manual id %q (an endpoint never adopts the host's learned UUID)", got, s.ID) + } + + // Regression: a classic address entry still re-keys to the learned UUID so + // a manually-added PAIR peer collapses onto its mDNS entry. + plain := manualNodeStatus{ + ID: "manual:10.0.1.10", + Address: "10.0.1.10", + OllamaUp: true, + OllamaPort: 11434, + NodeInfoPort: 14318, + HostUUID: "learned-host-uuid", + } + if got := manualToEnriched(plain).storeKey(); got != "learned-host-uuid" { + t.Fatalf("address-entry storeKey = %q, want the learned uuid", got) + } +} diff --git a/services/tests/broker_management_test.go b/services/tests/broker_management_test.go index 25a4afe7..74871d9c 100644 --- a/services/tests/broker_management_test.go +++ b/services/tests/broker_management_test.go @@ -143,6 +143,9 @@ func proxyNodesHas(t *testing.T, raw json.RawMessage, id string) bool { // shows up in proxy:nodes/list and can be routed to — even though it never // appears over mDNS. func TestBrokerBridgesManualNodeIntoProxy(t *testing.T) { + // node/add now persists via the manual-nodes service-owned store (appdir); + // point it at a temp dir so the test can't pollute the user's real store. + t.Setenv("XDG_CONFIG_HOME", t.TempDir()) if portBusy(11435) { t.Skip("ollama-proxy port 11435 already in use; skipping") } @@ -197,6 +200,8 @@ func TestBrokerBridgesManualNodeIntoProxy(t *testing.T) { // resolve to it. The candidate must therefore appear in proxy:nodes/list under // the learned hostUuid, not the user-supplied manual name. func TestBrokerBridgesManualNodeUnderLearnedUUID(t *testing.T) { + // Isolate the manual-nodes service-owned persistence (appdir). + t.Setenv("XDG_CONFIG_HOME", t.TempDir()) if portBusy(11435) { t.Skip("ollama-proxy port 11435 already in use; skipping") } @@ -254,6 +259,8 @@ func TestBrokerBridgesManualNodeUnderLearnedUUID(t *testing.T) { // into lmstudio-proxy (node/add-manual) so it shows up in // lmstudio-proxy:nodes/list — even though it never appears over mDNS. func TestBrokerBridgesManualNodeIntoLMStudioProxy(t *testing.T) { + // Isolate the manual-nodes service-owned persistence (appdir). + t.Setenv("XDG_CONFIG_HOME", t.TempDir()) configDir := persistLMStudioProxyPort(t, freePort(t)) stopLM := fakeLMStudio(t) // skips if 1234 unavailable t.Cleanup(stopLM) @@ -296,6 +303,78 @@ func TestBrokerBridgesManualNodeIntoLMStudioProxy(t *testing.T) { } } +// fakeOpenAIEndpoint serves a minimal OpenAI-compatible API at the ROOT of +// 127.0.0.1:port (no /v1 prefix — the declared base URL therefore carries an +// empty base path, which exercises the proxy's path join in both directions). +// It returns a stop function, and skips the test if the port is unavailable. +func fakeOpenAIEndpoint(t *testing.T, port int) func() { + t.Helper() + ln, err := net.Listen("tcp", fmt.Sprintf("127.0.0.1:%d", port)) + if err != nil { + t.Skipf("cannot bind fake OpenAI endpoint on 127.0.0.1:%d (%v); skipping", port, err) + } + mux := http.NewServeMux() + mux.HandleFunc("/models", func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + _, _ = io.WriteString(w, `{"object":"list","data":[{"id":"stub-model","object":"model"}]}`) + }) + srv := &http.Server{Handler: mux} + go func() { _ = srv.Serve(ln) }() + return func() { _ = srv.Close() } +} + +// TestBrokerBridgesOpenAIEndpointIntoLMStudioProxy: with the broker supervising +// both nvpair-manual-nodes and lmstudio-proxy, a manual node declared by an +// OpenAI base URL must be bridged into lmstudio-proxy (node/add-manual, +// carrying the endpoint's base_path) so it shows up in +// lmstudio-proxy:nodes/list — even though it never appears over mDNS. +func TestBrokerBridgesOpenAIEndpointIntoLMStudioProxy(t *testing.T) { + // node/add now persists via the manual-nodes service-owned store (appdir); + // point it at a temp dir so the test can't pollute the user's real store. + t.Setenv("XDG_CONFIG_HOME", t.TempDir()) + configDir := persistLMStudioProxyPort(t, freePort(t)) + endpointPort := freePort(t) + stopEP := fakeOpenAIEndpoint(t, endpointPort) + t.Cleanup(stopEP) + + stdin, msgs, _, cleanup := startBrokerWithConfigDir(t, configDir, + "--manual-nodes-path", manualNodesBin, + "--lmstudio-proxy-path", lmstudioProxyBin, + ) + t.Cleanup(cleanup) + + waitForMethod(t, msgs, "app:ready", 10*time.Second) + + const nodeName = "xproc-manual-openai-bridge" + addReq := fmt.Sprintf(`{"jsonrpc":"2.0","id":930,"method":"node/add","params":{"openai_base_url":"http://127.0.0.1:%d","name":%q}}`, endpointPort, nodeName) + "\n" + if _, err := stdin.Write([]byte(addReq)); err != nil { + t.Fatalf("write node/add: %v", err) + } + + deadline := time.After(25 * time.Second) + ticker := time.NewTicker(1 * time.Second) + defer ticker.Stop() + reqID := 931 + sendReq(t, stdin, reqID, "lmstudio-proxy:nodes/list") + for { + select { + case msg, ok := <-msgs: + if !ok { + t.Fatal("broker stream closed before the endpoint bridged into lmstudio-proxy") + } + if msg.Method == "" && msg.ID != nil && proxyNodesHas(t, msg.Result, nodeName) { + t.Logf("endpoint %q bridged into lmstudio-proxy nodes/list", nodeName) + return + } + case <-ticker.C: + reqID++ + sendReq(t, stdin, reqID, "lmstudio-proxy:nodes/list") + case <-deadline: + t.Fatalf("timed out waiting for endpoint %q in lmstudio-proxy:nodes/list", nodeName) + } + } +} + // TestBrokerAcceptsClientErrorsReport is the leg-B repro: a client-originated // errors:report (notification form) must be forwarded into nvpair-errors and // reflected in the next errors:update — the same as a supervised worker From 875216f4b3755f4782abb7ffe9196d6267d9fa0f Mon Sep 17 00:00:00 2001 From: Will Ford Date: Mon, 7 Sep 2026 23:16:08 +0200 Subject: [PATCH 3/7] lmstudio-proxy: forward OpenAI paths for base-prefixed endpoints External OpenAI-compatible endpoints adopted via a declared base URL (e.g. http://dgx:8000/v1) carry a base_path into the proxy. The /v1 prefix is the proxy's own API root, so inbound /v1/... paths are joined onto the endpoint's root instead of forwarded verbatim: model list fanout, the reverse-proxy Director, and the workload engine label (openai) all follow the endpoint's path. Nodes without a base path (discovered peers, classic manual nodes) are unchanged. Signed-off-by: Will Ford --- services/lmstudio-proxy/discovery.go | 7 ++ services/lmstudio-proxy/e2e_test.go | 122 ++++++++++++++++++++++++++ services/lmstudio-proxy/proxy.go | 75 +++++++++++++--- services/lmstudio-proxy/proxy_test.go | 20 +++++ 4 files changed, 211 insertions(+), 13 deletions(-) diff --git a/services/lmstudio-proxy/discovery.go b/services/lmstudio-proxy/discovery.go index 9b05581b..fb19e9c7 100644 --- a/services/lmstudio-proxy/discovery.go +++ b/services/lmstudio-proxy/discovery.go @@ -40,6 +40,12 @@ type Node struct { // advertises the requested model; an empty list stays in discovery but is // not an inference candidate until a later inventory update. Models []string `json:"models,omitempty"` + // BasePath is the endpoint's API path prefix (e.g. "/v1") for external + // OpenAI-compatible endpoints bridged with a declared base URL. The proxy + // joins it onto its own /v1 root when forwarding. Empty for every + // discovered node and for classic manual nodes, whose engines serve the + // OpenAI API at their own /v1 root. + BasePath string `json:"base_path,omitempty"` // IP is the single canonical LAN address a consumer should dial/display for // this node, resolved via the shared netpick ranker: the node's // own ip= TXT if present, else the best-scored advertised IPv4. It is @@ -135,6 +141,7 @@ func (d *Discovery) SetSubscribed(nodes []Node) (discovered, updated, removed [] // warrants a node/updated. func nodeEqual(a, b Node) bool { return a.ID == b.ID && a.Host == b.Host && a.Port == b.Port && a.IP == b.IP && + a.BasePath == b.BasePath && slices.Equal(a.Addresses, b.Addresses) && slices.Equal(a.TXT, b.TXT) && slices.Equal(a.Models, b.Models) } diff --git a/services/lmstudio-proxy/e2e_test.go b/services/lmstudio-proxy/e2e_test.go index 5412ab7c..dd6aa9ee 100644 --- a/services/lmstudio-proxy/e2e_test.go +++ b/services/lmstudio-proxy/e2e_test.go @@ -215,3 +215,125 @@ func TestE2EFailoverOverRealBinary(t *testing.T) { e2eSend(t, stdin, 9, "shutdown", nil) e2eWaitResult(t, frames, "9", 5*time.Second) } + +// TestE2EBasePathForwardingOverRealBinary spawns the real lmstudio-proxy +// binary and registers an external OpenAI endpoint whose API is rooted at +// /v1 (base_path "/v1"). It asserts the model list fanout and a genuine +// inference POST both land on the endpoint's /v1-rooted paths (not doubled +// /v1/v1/...), and that the workload is labeled engine "openai". +func TestE2EBasePathForwardingOverRealBinary(t *testing.T) { + var chatPath, modelsPath string + stub := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + switch { + case r.Method == http.MethodGet && r.URL.Path == "/v1/models": + modelsPath = r.URL.Path + io.WriteString(w, `{"object":"list","data":[{"id":"stub-model","object":"model"}]}`) + case r.Method == http.MethodPost && r.URL.Path == "/v1/chat/completions": + chatPath = r.URL.Path + io.Copy(io.Discard, r.Body) + io.WriteString(w, `{"id":"c1","object":"chat.completion","choices":[{"message":{"role":"assistant","content":"hi"}}]}`) + default: + w.WriteHeader(http.StatusNotFound) + io.WriteString(w, `{"error":"unexpected path `+r.URL.Path+`"}`) + } + })) + defer stub.Close() + + port := e2eFreePort(t) + cmd := exec.Command(proxyBin, "--port", strconv.Itoa(port)) + stdin, err := cmd.StdinPipe() + if err != nil { + t.Fatal(err) + } + stdout, err := cmd.StdoutPipe() + if err != nil { + t.Fatal(err) + } + if err := cmd.Start(); err != nil { + t.Fatal(err) + } + defer func() { + _ = stdin.Close() + _ = cmd.Process.Kill() + _ = cmd.Wait() + }() + + frames := make(chan e2eFrame, 256) + go e2eReadFrames(stdout, frames) + if got := e2eWaitReadyPort(t, frames, 10*time.Second); got != port { + t.Fatalf("ready port = %d, want %d", got, port) + } + + host, stubPort := e2eSplitHostPort(t, stub.URL) + e2eSend(t, stdin, 1, "node/add-manual", map[string]any{ + "id": "ep", "host": host, "port": stubPort, + "addresses": []string{host}, "models": []string{"stub-model"}, + "base_path": "/v1", + }) + e2eWaitResult(t, frames, "1", 5*time.Second) + + // Model list fanout must hit the endpoint's /v1/models. + list, err := http.Get(fmt.Sprintf("http://127.0.0.1:%d/v1/models", port)) + if err != nil { + t.Fatalf("model list GET: %v", err) + } + listBody, _ := io.ReadAll(list.Body) + list.Body.Close() + if list.StatusCode != http.StatusOK { + t.Fatalf("model list status = %d (body %s)", list.StatusCode, listBody) + } + if !strings.Contains(string(listBody), "stub-model") { + t.Fatalf("model list missing stub-model: %s", listBody) + } + if modelsPath != "/v1/models" { + t.Fatalf("endpoint served model list at %q, want /v1/models", modelsPath) + } + + // Inference POST must be joined onto the endpoint's /v1 root, verbatim. + resp, err := http.Post(fmt.Sprintf("http://127.0.0.1:%d/v1/chat/completions", port), "application/json", strings.NewReader(`{"model":"stub-model","messages":[{"role":"user","content":"hi"}]}`)) + if err != nil { + t.Fatalf("inference POST: %v", err) + } + respBody, _ := io.ReadAll(resp.Body) + resp.Body.Close() + if resp.StatusCode != http.StatusOK { + t.Fatalf("status = %d (body %s), want 200", resp.StatusCode, respBody) + } + if chatPath != "/v1/chat/completions" { + t.Fatalf("endpoint served chat at %q, want /v1/chat/completions (no /v1 doubling)", chatPath) + } + + // The workload must name the endpoint as the honest "openai" engine. + deadline := time.After(5 * time.Second) + gotWorkload := false + for !gotWorkload { + select { + case f := <-frames: + if f.Method != "workload:started" { + continue + } + var wp struct { + WorkloadInfo struct { + Engine string `json:"engine"` + ScheduledOn string `json:"scheduledOn"` + } `json:"workloadInfo"` + } + if err := json.Unmarshal(f.Params, &wp); err != nil { + t.Fatalf("parse workload:started params: %v", err) + } + if wp.WorkloadInfo.Engine != "openai" { + t.Fatalf("workload engine = %q, want openai", wp.WorkloadInfo.Engine) + } + if wp.WorkloadInfo.ScheduledOn != "ep" { + t.Fatalf("workload scheduledOn = %q, want ep", wp.WorkloadInfo.ScheduledOn) + } + gotWorkload = true + case <-deadline: + t.Fatal("timed out waiting for workload:started") + } + } + + e2eSend(t, stdin, 9, "shutdown", nil) + e2eWaitResult(t, frames, "9", 5*time.Second) +} diff --git a/services/lmstudio-proxy/proxy.go b/services/lmstudio-proxy/proxy.go index 6e619e77..f2eb0cd2 100644 --- a/services/lmstudio-proxy/proxy.go +++ b/services/lmstudio-proxy/proxy.go @@ -19,8 +19,10 @@ import ( "net/http" "net/http/httputil" "net/url" + "path" "sort" "strconv" + "strings" "sync" "sync/atomic" "time" @@ -141,8 +143,23 @@ const ( // workloadEngine is the opaque engine identifier carried in every // workload this proxy produces. This proxy only ever fronts LM Studio. workloadEngine = "lmstudio" + + // externalEndpointEngine labels workloads served by an external + // OpenAI-compatible endpoint (a base-path manual node), which is not LM + // Studio — the workload's engine must name what actually served it. + externalEndpointEngine = "openai" ) +// engineForCandidate returns the workload engine label for a forwarding +// candidate: external endpoints label "openai", everything else is this +// proxy's LM Studio family. +func engineForCandidate(basePath string) string { + if basePath != "" { + return externalEndpointEngine + } + return workloadEngine +} + // inferenceEndpoints is the set of request paths that count as cluster // workloads. Health checks, model listings (/v1/models), and other control // traffic are deliberately excluded so we don't flood the cluster with @@ -700,6 +717,10 @@ type candidate struct { id string url *url.URL peerUUID string + // basePath is the endpoint's API prefix (e.g. "/v1") carried by the node + // for external OpenAI-compatible endpoints; "" means the OpenAI paths + // forward verbatim. + basePath string } // candidateTransport returns the reverse-proxy / model-list transport for a @@ -825,8 +846,10 @@ func (p *Proxy) serveModelList(w http.ResponseWriter, r *http.Request, candidate var wg sync.WaitGroup for i, cand := range candidates { target := *cand.url - target.Path = r.URL.Path - target.RawPath = r.URL.RawPath + // The endpoint's own /models lives under its base path, so the fanout + // uses the same join the forward path does. + target.Path = joinedPath(cand.basePath, r.URL.Path) + target.RawPath = "" target.RawQuery = r.URL.RawQuery upstream, err := http.NewRequestWithContext(r.Context(), http.MethodGet, target.String(), nil) if err != nil { @@ -1042,7 +1065,7 @@ func (p *Proxy) handleHTTP(w http.ResponseWriter, r *http.Request) { wl = &Workload{ ID: reqID, Model: model, - Engine: workloadEngine, + Engine: engineForCandidate(candidates[0].basePath), RunID: p.runID, State: "running", ScheduledOn: candidates[0].id, @@ -1134,6 +1157,13 @@ func (p *Proxy) handleHTTP(w http.ResponseWriter, r *http.Request) { req.URL.Scheme = cand.url.Scheme req.URL.Host = cand.url.Host req.Host = cand.url.Host + // External endpoints rooted at a base path (e.g. /v1) receive + // the proxy's /v1-prefixed path rewritten onto their root. + // Clear RawPath so the rewritten Path is what goes on the wire. + if cand.basePath != "" { + req.URL.Path = joinedPath(cand.basePath, req.URL.Path) + req.URL.RawPath = "" + } }, // A remote cluster peer is dialed over mTLS (per-peer pinned config); // self/manual candidates use the plain transport. See candidateTransport. @@ -1190,18 +1220,21 @@ func (p *Proxy) handleHTTP(w http.ResponseWriter, r *http.Request) { // the node that actually served. Guarded by wlMu against the // disconnect watcher, and skipped once terminated so a late // re-point can't resurrect a workload we've already failed. - if wl != nil { - wlMu.Lock() - if !terminated && wl.ScheduledOn != cand.id { - wl.ScheduledOn = cand.id - snapshot := *wl - wlMu.Unlock() - p.emitWorkload(workloadStartedMethod, snapshot) - } else { - wlMu.Unlock() - } + if wl != nil { + wlMu.Lock() + if !terminated && wl.ScheduledOn != cand.id { + wl.ScheduledOn = cand.id + // Failover can land on (or off) an external endpoint; + // the engine label must follow the node that serves. + wl.Engine = engineForCandidate(cand.basePath) + snapshot := *wl + wlMu.Unlock() + p.emitWorkload(workloadStartedMethod, snapshot) + } else { + wlMu.Unlock() } } + } return nil }, ErrorHandler: func(ew http.ResponseWriter, _ *http.Request, err error) { @@ -1433,6 +1466,7 @@ func (p *Proxy) resolveCandidates(model string) []candidate { id: n.ID, url: u, peerUUID: peerUUID, + basePath: n.BasePath, }) } @@ -1556,6 +1590,21 @@ func isSelfTarget(u *url.URL, selfPort int) bool { return false } +// joinedPath rewrites the proxy's /v1-prefixed request path for a candidate +// whose API is rooted at basePath (e.g. "/v1"): the inbound path's "/v1" +// prefix is the proxy's own API root and is replaced by the endpoint's. +// Candidates without a base path, and inbound paths outside /v1 (control +// traffic), pass through unchanged. +func joinedPath(basePath, inbound string) string { + if basePath == "" { + return inbound + } + if !strings.HasPrefix(inbound, "/v1/") { + return inbound + } + return path.Join(basePath, strings.TrimPrefix(inbound, "/v1")) +} + // nodeURL returns the single best forward URL for a node (the first candidate // in deterministic, loopback-first order). It does no reachability probing — // p.targetURL is the request-path entry point. Kept as a free function so the diff --git a/services/lmstudio-proxy/proxy_test.go b/services/lmstudio-proxy/proxy_test.go index d0a029c6..a075d2d0 100644 --- a/services/lmstudio-proxy/proxy_test.go +++ b/services/lmstudio-proxy/proxy_test.go @@ -61,3 +61,23 @@ func TestNodeURL(t *testing.T) { }) } } + +func TestJoinedPath(t *testing.T) { + cases := []struct { + base, in, want string + }{ + {"", "/v1/chat/completions", "/v1/chat/completions"}, + {"/v1", "/v1/chat/completions", "/v1/chat/completions"}, + {"/v1", "/v1/models", "/v1/models"}, + {"/v1", "/v1/completions", "/v1/completions"}, + {"/api", "/v1/embeddings", "/api/embeddings"}, + {"/v1", "/health", "/health"}, + {"", "/health", "/health"}, + {"/nested/v1", "/v1/chat/completions", "/nested/v1/chat/completions"}, + } + for _, tc := range cases { + if got := joinedPath(tc.base, tc.in); got != tc.want { + t.Errorf("joinedPath(%q, %q) = %q, want %q", tc.base, tc.in, got, tc.want) + } + } +} From 8a59b1a1366734830aae19b3ec43c71950b8f41f Mon Sep 17 00:00:00 2001 From: Will Ford Date: Mon, 7 Sep 2026 23:16:38 +0200 Subject: [PATCH 4/7] frontends: add OpenAI endpoints by URL (TUI + desktop) The TUI add prompt accepts a full http:// base URL in addition to a bare host (with an OPENAI reachability column). The desktop Add-node dialog gains an 'OpenAI endpoint' row that relays node/add with openai_base_url (nvpair-manual-nodes persists the entry). Workload engine 'openai' maps onto the existing lm-studio display path so external-endpoint jobs are not dropped by the closed union. Signed-off-by: Will Ford --- .../electron/service-bridge/empty-handlers.ts | 26 +++++++ .../electron/service-bridge/modular-state.ts | 8 ++- desktop/src/shared/types/ws-channels.ts | 6 ++ desktop/src/ui/api/pair-api.ts | 3 + desktop/src/ui/components/AddNodeModal.tsx | 58 ++++++++++++++++ .../tests/modular/openai-endpoint-add.test.ts | 67 +++++++++++++++++++ .../modular/openai-engine-workload.test.ts | 60 +++++++++++++++++ services/nvpair-tui/ui/manualnodes.go | 55 +++++++++++---- services/nvpair-tui/ui/manualnodes_test.go | 37 ++++++++++ 9 files changed, 305 insertions(+), 15 deletions(-) create mode 100644 desktop/tests/modular/openai-endpoint-add.test.ts create mode 100644 desktop/tests/modular/openai-engine-workload.test.ts create mode 100644 services/nvpair-tui/ui/manualnodes_test.go diff --git a/desktop/src/electron/service-bridge/empty-handlers.ts b/desktop/src/electron/service-bridge/empty-handlers.ts index dc5c2dd8..68f76b63 100644 --- a/desktop/src/electron/service-bridge/empty-handlers.ts +++ b/desktop/src/electron/service-bridge/empty-handlers.ts @@ -944,6 +944,31 @@ async function handleNodeRemoveMember( } } +/** + * Adopt an externally-managed OpenAI-compatible endpoint by base URL. + * Relays `node/add` with `openai_base_url` to the broker, which forwards it to + * nvpair-manual-nodes (persisting the entry and probing the declared URL). + * The scheme check stays local for fast inline feedback; the service + * re-validates and is authoritative. + */ +async function handleAddOpenAIEndpoint( + payload?: WsInvokeRequest<'nodes:add-endpoint'> +): Promise> { + const url = payload?.url?.trim() ?? '' + if (!/^https?:\/\//.test(url)) { + return { + ok: false, + error: 'Enter a full endpoint URL, e.g. http://192.168.1.50:8888/v1 (https not supported yet)' + } + } + try { + await getModularSupervisor().callProcess('broker', 'node/add', { openai_base_url: url }) + return { ok: true } + } catch (err) { + return { ok: false, error: getErrorString(err) } + } +} + const EMPTY_SERVICE_BRIDGE_HANDLERS: BridgeHandlerMap = { 'app:get-initial': async () => ({ connected: getModularSupervisor().ready, @@ -952,6 +977,7 @@ const EMPTY_SERVICE_BRIDGE_HANDLERS: BridgeHandlerMap = { 'nodes:get-initial': () => getModularBridgeState().getNodesInitial(), 'nodes:remove-member': payload => handleNodeRemoveMember(payload), + 'nodes:add-endpoint': payload => handleAddOpenAIEndpoint(payload), 'discovery:get-nodes': () => getModularBridgeState().getAvailableNodes(), diff --git a/desktop/src/electron/service-bridge/modular-state.ts b/desktop/src/electron/service-bridge/modular-state.ts index ab6c5fb5..e6c62adc 100644 --- a/desktop/src/electron/service-bridge/modular-state.ts +++ b/desktop/src/electron/service-bridge/modular-state.ts @@ -403,10 +403,14 @@ function pendingEngineOpIdleTimeoutMs(status: EngineProcessStatus): number { /** * Map a `nvpair-engine-manager` engine identifier onto our closed `EngineType` - * union. The engine-manager uses `lmstudio`; we use `lm-studio`. + * union. The engine-manager uses `lmstudio`; we use `lm-studio`. External + * OpenAI-compatible endpoints report engine `openai`; they render through the + * existing lm-studio display path (no dedicated icon/badge — cosmetic, same as + * the node-card engine chip) so their workloads are not silently dropped by + * the closed-union guard. */ function engineManagerEngineType(name: string): EngineType | null { - const normalized = name === 'lmstudio' ? 'lm-studio' : name + const normalized = name === 'lmstudio' || name === 'openai' ? 'lm-studio' : name return isEngineType(normalized) ? normalized : null } diff --git a/desktop/src/shared/types/ws-channels.ts b/desktop/src/shared/types/ws-channels.ts index 07398eab..cf3aa439 100644 --- a/desktop/src/shared/types/ws-channels.ts +++ b/desktop/src/shared/types/ws-channels.ts @@ -69,6 +69,12 @@ export interface WsInvokeChannelMap { request: { nodeId: string } response: { nodeId: string; removed: boolean } } + // Adopt an externally-managed OpenAI-compatible endpoint by base URL + // (relayed to the broker's `node/add`; nvpair-manual-nodes persists it). + 'nodes:add-endpoint': { + request: { url: string } + response: { ok: boolean; error?: string } + } // Discovery 'discovery:get-nodes': { request: void; response: AvailableNode[] } diff --git a/desktop/src/ui/api/pair-api.ts b/desktop/src/ui/api/pair-api.ts index a0a34155..3d0b49ed 100644 --- a/desktop/src/ui/api/pair-api.ts +++ b/desktop/src/ui/api/pair-api.ts @@ -39,6 +39,8 @@ export interface INodesApi { }> /** Remove a node from the cluster (revokes membership + pinned trust). */ removeMember(nodeId: string): Promise<{ nodeId: string; removed: boolean }> + /** Adopt an externally-managed OpenAI-compatible endpoint by base URL. */ + addEndpoint(url: string): Promise<{ ok: boolean; error?: string }> /** A node was added or updated in the discovery/metrics list. */ onUpsert(callback: (node: NodeItem) => void): () => void /** A node was removed from the discovery/metrics list. */ @@ -146,6 +148,7 @@ export function createPairApi(transport: ServiceTransport): IPairApi { } }, removeMember: nodeId => transport.invoke('nodes:remove-member', { nodeId }), + addEndpoint: url => transport.invoke('nodes:add-endpoint', { url }), onUpsert: cb => transport.subscribePush('nodes:upsert', cb), onRemove: cb => transport.subscribePush('nodes:remove', cb), onMembersChanged: cb => transport.subscribePush('nodes:changed', cb) diff --git a/desktop/src/ui/components/AddNodeModal.tsx b/desktop/src/ui/components/AddNodeModal.tsx index d54abde9..961ef6c6 100644 --- a/desktop/src/ui/components/AddNodeModal.tsx +++ b/desktop/src/ui/components/AddNodeModal.tsx @@ -16,6 +16,7 @@ import { } from '@nvidia/foundations-react-core' import { DialogHeader } from './DialogHeader' import { InvitePairingPanel } from './InvitePairingPanel' +import getErrorString from '@/shared/utils/get-error-string' import { useBlurOnOpen } from '@/ui/hooks/useBlurOnOpen' import { useInvitePairing } from '@/ui/hooks/useInvitePairing' import { useInvitablePeers } from '@/ui/hooks/useInvitablePeers' @@ -28,12 +29,17 @@ interface AddNodeModalProps { export function AddNodeModal({ open, onOpenChange }: AddNodeModalProps) { useBlurOnOpen(open) const [manualIp, setManualIp] = useState('') + const [endpointUrl, setEndpointUrl] = useState('') + const [endpointError, setEndpointError] = useState(null) + const [endpointInFlight, setEndpointInFlight] = useState(false) const pairing = useInvitePairing() const nodesThatCanBeAdded = useInvitablePeers() const handleOpenChange = useCallback( (next: boolean) => { setManualIp('') + setEndpointUrl('') + setEndpointError(null) pairing.reset() onOpenChange(next) }, @@ -46,6 +52,25 @@ export function AddNodeModal({ open, onOpenChange }: AddNodeModalProps) { void pairing.start(ip) }, [manualIp, pairing]) + const handleAddEndpoint = useCallback(async () => { + const url = endpointUrl.trim() + if (!url || endpointInFlight) return + setEndpointInFlight(true) + setEndpointError(null) + try { + const result = await window.pairApi.nodes.addEndpoint(url) + if (result.ok) { + handleOpenChange(false) + } else { + setEndpointError(result.error ?? 'Failed to add the endpoint.') + } + } catch (err) { + setEndpointError(getErrorString(err)) + } finally { + setEndpointInFlight(false) + } + }, [endpointUrl, endpointInFlight, handleOpenChange]) + const showPairing = pairing.invite !== null || pairing.error !== null const inviteInFlight = pairing.submitting || pairing.invite?.state === 'pending' @@ -94,6 +119,39 @@ export function AddNodeModal({ open, onOpenChange }: AddNodeModalProps) { + + + + + { + setEndpointUrl(value) + setEndpointError(null) + }} + placeholder="http://192.168.1.50:8888/v1" + onKeyDown={event => { + if (event.key === 'Enter') { + void handleAddEndpoint() + } + }} + disabled={endpointInFlight} + /> + + + + {endpointError && ( + {endpointError} + )} + + {nodesThatCanBeAdded.length > 0 && ( diff --git a/desktop/tests/modular/openai-endpoint-add.test.ts b/desktop/tests/modular/openai-endpoint-add.test.ts new file mode 100644 index 00000000..ab61e973 --- /dev/null +++ b/desktop/tests/modular/openai-endpoint-add.test.ts @@ -0,0 +1,67 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +import { describe, expect, it, vi } from 'vitest' + +const mocks = vi.hoisted(() => ({ + state: { + getSelfId: vi.fn(() => 'local-node') + }, + supervisor: { + callProcess: vi.fn(), + sendProcess: vi.fn(), + reportError: vi.fn() + } +})) + +vi.mock('@/electron/service-bridge/modular-supervisor', () => ({ + getModularSupervisor: () => mocks.supervisor +})) +vi.mock('@/electron/service-bridge/modular-state', () => ({ + getModularBridgeState: () => mocks.state, + isProxyEngine: () => false, + isUpstreamUnreachableError: () => false, + parseServiceErrors: () => [], + parseWorkloadsInitial: () => [] +})) +vi.mock('@/electron/model-hub', () => ({ getEngineHubModels: vi.fn() })) + +import { handleServiceBridgeInvoke } from '@/electron/service-bridge/empty-handlers' + +describe('nodes:add-endpoint (adopt an external OpenAI endpoint by URL)', () => { + it('rejects a URL without a scheme before touching the service', async () => { + const result = await handleServiceBridgeInvoke('nodes:add-endpoint', { + url: 'localhost:8888' + }) + + expect(result.ok).toBe(false) + expect(result.error).toMatch(/http:\/\//) + expect(mocks.supervisor.callProcess).not.toHaveBeenCalled() + }) + + it('relays a full http URL to the broker as node/add with openai_base_url', async () => { + mocks.supervisor.callProcess.mockResolvedValue(undefined) + + const result = await handleServiceBridgeInvoke('nodes:add-endpoint', { + url: ' http://192.168.1.50:8888/v1 ' + }) + + expect(mocks.supervisor.callProcess).toHaveBeenCalledWith('broker', 'node/add', { + openai_base_url: 'http://192.168.1.50:8888/v1' + }) + expect(result).toEqual({ ok: true }) + }) + + it('surfaces a service failure as an inline error, not a thrown rejection', async () => { + mocks.supervisor.callProcess.mockRejectedValue( + new Error('node already registered for this endpoint') + ) + + const result = await handleServiceBridgeInvoke('nodes:add-endpoint', { + url: 'http://192.168.1.50:8888/v1' + }) + + expect(result.ok).toBe(false) + expect(result.error).toContain('node already registered') + }) +}) diff --git a/desktop/tests/modular/openai-engine-workload.test.ts b/desktop/tests/modular/openai-engine-workload.test.ts new file mode 100644 index 00000000..3dadfcb6 --- /dev/null +++ b/desktop/tests/modular/openai-engine-workload.test.ts @@ -0,0 +1,60 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +import { describe, expect, it, vi } from 'vitest' + +vi.mock('electron', () => ({ BrowserWindow: { getAllWindows: () => [] } })) +vi.mock('@/electron/window', () => ({ createOverviewWindow: vi.fn() })) + +import { parseWorkloadsInitial } from '@/electron/service-bridge/modular-state' + +// External OpenAI-compatible endpoints report the workload engine "openai". +// The desktop's EngineType union is closed; the mapping must render those +// workloads through the existing lm-studio path instead of silently dropping +// them at the unknown-engine guard. +describe('workload engine mapping for external endpoints', () => { + it('keeps an "openai" workload, rendered as lm-studio', () => { + const workloads = parseWorkloadsInitial({ + workloads: [ + { + id: 'job-openai', + engine: 'openai', + state: 'running', + model: 'stub-model', + originatedFrom: 'openai-wl-seed', + createdAt: 100 + } + ] + }) + + expect(workloads).toHaveLength(1) + expect(workloads[0].engine).toBe('lm-studio') + }) + + it('still maps the engine-manager "lmstudio" id and drops unknown engines', () => { + const workloads = parseWorkloadsInitial({ + workloads: [ + { + id: 'job-lmstudio', + engine: 'lmstudio', + state: 'running', + model: 'local-model', + originatedFrom: 'openai-wl-seed', + createdAt: 100 + }, + { + id: 'job-mystery', + engine: 'mystery-engine', + state: 'running', + model: 'ghost-model', + originatedFrom: 'openai-wl-seed', + createdAt: 200 + } + ] + }) + + expect(workloads).toHaveLength(1) + expect(workloads[0].id).toBe('job-lmstudio') + expect(workloads[0].engine).toBe('lm-studio') + }) +}) diff --git a/services/nvpair-tui/ui/manualnodes.go b/services/nvpair-tui/ui/manualnodes.go index fa316806..feccfcaf 100644 --- a/services/nvpair-tui/ui/manualnodes.go +++ b/services/nvpair-tui/ui/manualnodes.go @@ -4,6 +4,7 @@ package ui import ( + "fmt" "strings" "time" @@ -27,6 +28,7 @@ type manualNode struct { Address string `json:"address"` OllamaUp bool `json:"ollama_up"` NodeInfoUp bool `json:"node_info_up"` + OpenAIUp bool `json:"openai_up"` } // manualView manages user-added nodes that don't appear via mDNS: list, @@ -61,10 +63,11 @@ var ( func newManualView(client *rpc.Client) *manualView { ti := textinput.New() - // nvpair-manual-nodes appends its own fixed ports (11434 for Ollama, 14318 - // for node-info) to whatever's entered, so a host:port form yields a - // malformed URL and the node always reads down. Only a bare host works. - ti.Placeholder = "host" + // A bare host uses the historical form: nvpair-manual-nodes appends its + // fixed engine ports (11434 for Ollama, 14318 for node-info), so a + // host:port form yields a malformed URL and the node always reads down. + // A full http:// base URL adopts the endpoint at that exact URL instead. + ti.Placeholder = "host or http://host:port/v1" v := &manualView{client: client, input: ti} v.table = newTable(nil) return v @@ -95,14 +98,15 @@ func (v *manualView) tickCmd() tea.Cmd { func (v *manualView) SetSize(w, h int) { v.width, v.height = w, h - const ollama, nodeinfo = 8, 9 - id := clampWidth((w-ollama-nodeinfo-2)/2, 8) - addr := clampWidth(w-ollama-nodeinfo-id-2, 10) + const ollama, nodeinfo, openai = 8, 9, 8 + id := clampWidth((w-ollama-nodeinfo-openai-2)/2, 8) + addr := clampWidth(w-ollama-nodeinfo-openai-id-2, 10) v.table.SetColumns([]table.Column{ {Title: "ID", Width: id}, {Title: "ADDRESS", Width: addr}, {Title: "OLLAMA", Width: ollama}, {Title: "NODEINFO", Width: nodeinfo}, + {Title: "OPENAI", Width: openai}, }) v.table.SetWidth(w) v.table.SetHeight(clampWidth(h-2, 1)) @@ -160,16 +164,40 @@ func (v *manualView) handleKey(msg tea.KeyMsg) tea.Cmd { return cmd } +// parseManualInput splits the operator's add input into the two node/add +// flavors: a bare host/address (the historical form; the prober appends its +// fixed engine ports) or a full OpenAI-compatible base URL (declared +// endpoints are probed at the URL itself, not the default ports). +func parseManualInput(s string) (address, baseURL string, err error) { + s = strings.TrimSpace(s) + if strings.Contains(s, "://") { + if !strings.HasPrefix(s, "http://") { + return "", "", fmt.Errorf("only http:// base URLs are supported (https not yet)") + } + return "", s, nil + } + if s == "" { + return "", "", fmt.Errorf("address required") + } + return s, "", nil +} + func (v *manualView) submitAdd() tea.Cmd { v.adding = false v.input.Blur() - addr := strings.TrimSpace(v.input.Value()) - if addr == "" { - v.status = "address required" + addr, baseURL, err := parseManualInput(v.input.Value()) + if err != nil { + v.status = err.Error() return nil } - return call(v.client, "node/add", map[string]string{"address": addr}, func(_ *rpc.Message, err error) tea.Msg { - return manualActionMsg{what: "add " + addr, err: err} + params := map[string]string{"address": addr} + label := addr + if baseURL != "" { + params = map[string]string{"openai_base_url": baseURL} + label = baseURL + } + return call(v.client, "node/add", params, func(_ *rpc.Message, err error) tea.Msg { + return manualActionMsg{what: "add " + label, err: err} }) } @@ -193,6 +221,7 @@ func (v *manualView) setNodes(nodes []manualNode) { n.Address, yesNo(n.OllamaUp), yesNo(n.NodeInfoUp), + yesNo(n.OpenAIUp), }) } v.table.SetRows(rows) @@ -201,7 +230,7 @@ func (v *manualView) setNodes(nodes []manualNode) { func (v *manualView) View() string { var b strings.Builder if len(v.nodes) == 0 { - b.WriteString(footerStyle.Render("No manual nodes. Press a to add one by address.")) + b.WriteString(footerStyle.Render("No manual nodes. Press a to add one by host or OpenAI endpoint URL.")) } else { b.WriteString(v.table.View()) } diff --git a/services/nvpair-tui/ui/manualnodes_test.go b/services/nvpair-tui/ui/manualnodes_test.go new file mode 100644 index 00000000..55c9094d --- /dev/null +++ b/services/nvpair-tui/ui/manualnodes_test.go @@ -0,0 +1,37 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +package ui + +import ( + "testing" +) + +func TestParseManualInput(t *testing.T) { + cases := []struct { + in string + address string + baseURL string + wantErr bool + }{ + {in: "192.168.1.50", address: "192.168.1.50"}, + {in: "dgx", address: "dgx"}, + {in: "http://localhost:8888/v1", baseURL: "http://localhost:8888/v1"}, + {in: "http://dgx:8000", baseURL: "http://dgx:8000"}, + {in: "https://localhost:8888/v1", wantErr: true}, // http only, say so + {in: " ", wantErr: true}, + } + for _, tc := range cases { + addr, base, err := parseManualInput(tc.in) + if tc.wantErr { + if err == nil { + t.Errorf("parseManualInput(%q): expected error", tc.in) + } + continue + } + if err != nil || addr != tc.address || base != tc.baseURL { + t.Errorf("parseManualInput(%q) = %q %q %v, want %q %q", + tc.in, addr, base, err, tc.address, tc.baseURL) + } + } +} From f16aeca5f7c6224666155504d169354bdf00c347 Mon Sep 17 00:00:00 2001 From: Will Ford Date: Mon, 7 Sep 2026 23:17:21 +0200 Subject: [PATCH 5/7] docs: OpenAI endpoint adoption (manual-nodes + proxy READMEs, versions) Document the openai_base_url entry flavor, service-owned entry persistence, the proxy's base_path forwarding, and the endpoint identity rule. Bump manual-nodes 0.12.0, ui-broker 0.41.0, lmstudio-proxy 0.17.0, tui 0.8.0, product 0.92.0. Signed-off-by: Will Ford --- docs/architecture.mdx | 9 +++++--- services/lmstudio-proxy/README.md | 4 ++++ services/nvpair-manual-nodes/README.md | 32 +++++++++++++++++++++----- services/versions.json | 12 +++++----- 4 files changed, 42 insertions(+), 15 deletions(-) diff --git a/docs/architecture.mdx b/docs/architecture.mdx index f7e7ffed..6b8b4736 100644 --- a/docs/architecture.mdx +++ b/docs/architecture.mdx @@ -648,9 +648,12 @@ its address. Some networks block or filter multicast, so discovery is not the only path in. `nvpair-manual-nodes` takes an address you enter directly and probes it on a -fixed interval, and a manual node that answers is folded into the same directory -as a discovered one. It is initially keyed by the address you typed, and re-keyed -to the peer's real UUID as soon as that node reports it. +fixed interval, or an OpenAI-compatible endpoint you declare by base URL and +probes at that exact URL, and a manual node that answers is folded into the +same directory as a discovered one. It is initially keyed by the address you +typed, and re-keyed to the peer's real UUID as soon as that node reports it. +The service owns its entry list in the application data directory and restores +it at startup, so manual nodes survive a restart. ## Trust Boundaries diff --git a/services/lmstudio-proxy/README.md b/services/lmstudio-proxy/README.md index 71a8b70d..747b77b9 100644 --- a/services/lmstudio-proxy/README.md +++ b/services/lmstudio-proxy/README.md @@ -93,6 +93,7 @@ Nodes are represented throughout the protocol with this shape: | `txt` | string[] | The discovery record's TXT pairs, carried verbatim | | `models` | string[] | The node's LM Studio model inventory from the discovery snapshot. Model-bearing inference is eligible only when this list advertises the exact requested model ID. An omitted or empty list excludes the node from that request until inventory updates; it remains available for non-inference routes and model-list aggregation | | `ip` | string | The single canonical LAN address to dial or display, resolved from the node's `ip=` TXT if present and otherwise the best-scored advertised IPv4. Stamped onto outbound `node/*` notifications so consumers agree with the address the proxy routes to | +| `base_path` | string | API path prefix (e.g. `/v1`) carried by an external OpenAI-compatible endpoint bridged with a declared base URL. The proxy joins it onto its own `/v1` root when forwarding requests and when fetching the node's model list. Absent for discovered nodes and for classic manual nodes, whose API is served at their own `/v1` root and forwards verbatim | --- @@ -350,6 +351,7 @@ Add a node manually (for networks where mDNS is blocked). If the node ID already **Request:** ```json {"jsonrpc":"2.0","id":5,"method":"node/add-manual","params":{"id":"remote-server","host":"remote-server","port":1234,"addresses":["10.0.1.50"]}} +{"jsonrpc":"2.0","id":5,"method":"node/add-manual","params":{"id":"vllm-host","host":"vllm-host","port":8888,"addresses":["192.168.1.50"],"models":["llama3.1:8b"],"base_path":"/v1"}} ``` **Response:** @@ -359,6 +361,8 @@ Add a node manually (for networks where mDNS is blocked). If the node ID already The proxy emits a `node/discovered` notification (or `node/updated` if the node was already registered). Manual nodes are a separate overlay that discovery snapshots never touch — they persist until explicitly removed. +A manual node carrying `base_path` is an external OpenAI-compatible endpoint adopted by declared base URL: the proxy rewrites its `/v1`-prefixed paths onto the endpoint's root (see the Node Object) and labels the workloads it serves with `engine: "openai"`, since the serving software is not LM Studio. + #### `node/remove-manual` Remove a previously added manual node. diff --git a/services/nvpair-manual-nodes/README.md b/services/nvpair-manual-nodes/README.md index 5fb648a6..2df51325 100644 --- a/services/nvpair-manual-nodes/README.md +++ b/services/nvpair-manual-nodes/README.md @@ -5,7 +5,7 @@ SPDX-License-Identifier: Apache-2.0 # nvpair-manual-nodes -A Go service for managing manually configured nodes on networks where mDNS discovery is unavailable. Accepts node addresses via JSON-RPC, probes each for Ollama, LM Studio, and node-info, and emits status events. +A Go service for managing manually configured nodes on networks where mDNS discovery is unavailable. Accepts node addresses or OpenAI-compatible endpoint URLs via JSON-RPC, probes each, and emits status events. ## Communication @@ -62,6 +62,8 @@ Emitted when a manually added node has been probed and its initial status determ Each node is probed for both inference engines: Ollama on its default `:11434` (`GET /` + `/api/tags`) and LM Studio on its default `:1234` (`GET /v1/models`, which doubles as the liveness check and the model list). `lmstudio_up` / `lmstudio_port` / `lmstudio_models` mirror the `ollama_*` fields and let a supervising broker bridge the node into `lmstudio-proxy` the same way it bridges Ollama into `ollama-proxy`. A node can run either engine, both, or neither. +For an entry added with `openai_base_url`, the `openai_*` fields describe the declared endpoint instead: `openai_up` is liveness, `openai_base_url` echoes the declared URL, `openai_host` / `openai_port` / `openai_base_path` are the parsed parts a supervising broker hands to the proxy (host for dialing, path prefix for forwarding), and `openai_models` is the inventory fetched from the endpoint. Address-based entries keep every `openai_*` field empty/false. + ### `node/updated` Emitted when a periodic probe detects a change (service going up/down, models, @@ -81,21 +83,28 @@ Emitted so the supervising broker can forward them into the `nvpair-errors` pipe ### `node/add` -Add a node by address. The manager immediately probes it and emits a `node/discovered` event. +Add a node by address or by OpenAI-compatible endpoint URL. The manager immediately probes it and emits a `node/discovered` event. + +Exactly one of `address` / `openai_base_url` must be supplied: -A hostname is preferred over an IP literal: probe clients disable keep-alives specifically so every probe re-resolves the name, which lets a node that gets a new address recover on its own. An IP-literal entry is dead once the device is renumbered. Supply the address on its own — a `host:port` string is not parsed, because ports are appended to it, so such an entry reads permanently down. +- `address` names a host the manager probes on the default engine ports (see [Probing](#probing)). A hostname is preferred over an IP literal: probe clients disable keep-alives specifically so every probe re-resolves the name, which lets a node that gets a new address recover on its own. An IP-literal entry is dead once the device is renumbered. Supply the address on its own — a `host:port` string is not parsed, because ports are appended to it, so such an entry reads permanently down. +- `openai_base_url` declares an externally-managed OpenAI-compatible endpoint at that exact URL (see [OpenAI-compatible endpoints](#openai-compatible-endpoints)). ```json {"jsonrpc":"2.0","id":1,"method":"node/add","params":{"address":"10.0.1.50","name":"my-server"}} +{"jsonrpc":"2.0","id":1,"method":"node/add","params":{"openai_base_url":"http://192.168.1.50:8888/v1"}} ``` | Param | Required | Description | |---|---|---| -| `address` | Yes | IP address or hostname of the node, with no port | -| `name` | No | Friendly name (used as node ID; defaults to `manual:
`) | +| `address` | Yes* | IP address or hostname of the node, with no port | +| `openai_base_url` | Yes* | OpenAI-compatible endpoint base URL, e.g. `http://192.168.1.50:8888/v1`. http only in this release | +| `name` | No | Friendly name (used as node ID; defaults to `manual:
`, or `manual::` for an endpoint) | | `tls_port` | No | Probe node-info over HTTPS on this port instead of plain HTTP on `14318`. Echoed back as `tls_enabled` | | `mtls` | No | Stored and echoed back as `mtls_required`. The probe transport itself is chosen by `tls_port` and live cluster membership, so this field records intent rather than driving it | +\*Exactly one of `address` / `openai_base_url` is required; supplying both (or neither) is rejected. + Response: the initial node status object. ### `node/remove` @@ -140,7 +149,18 @@ Each manual node is probed every 10 seconds, with a 3-second timeout per leg, fo A node can have any combination of these, or none if the target is unreachable. Status changes trigger `node/updated` events. Because change detection compares CPU, memory, and GPU values, a node running node-info emits a `node/updated` on most probe cycles as utilization moves. -The three engine ports are compiled in: only the node-info leg's port can be moved, via `tls_port`. A remote engine on a non-default port is not discovered. +The default engine ports are compiled in: only the node-info leg's port can be moved, via `tls_port`. An address-based entry that runs a remote engine on a non-default port is not discovered — but the same engine remains reachable by declaring its exact URL as an OpenAI-compatible endpoint. + +## OpenAI-compatible endpoints + +An entry added with `openai_base_url` adopts an externally-managed OpenAI-compatible API (any server that speaks the OpenAI HTTP API — vLLM, llama.cpp server, TGI, and the like) without the manager needing to know which software sits behind it. For this entry type the manager probes: + +- **The endpoint itself**: `GET {base}/models` — liveness check and model list in one (`openai_up`, `openai_models`) +- **Node Info**, best effort: on the URL's host, on the default `14318` (or `tls_port`), for hardware inventory and identity so the node folds into the same directory as a discovered one + +The default engine-port legs (Ollama `:11434`, LM Studio `:1234`) are not probed for this entry type; their status fields stay false/empty. + +The manager persists its entry list in the application data directory (`manual-nodes.json`) on every add and remove and restores it at startup, so manual nodes — declared endpoints included — survive a service restart. ## Shutdown diff --git a/services/versions.json b/services/versions.json index 29d8c230..20490cc9 100644 --- a/services/versions.json +++ b/services/versions.json @@ -1,20 +1,20 @@ { "$comment": "Single source of truth for all version numbers. See VERSIONING.md for bump rules.", - "product": "0.91.7", - "installer": "0.91.7", + "product": "0.92.0", + "installer": "0.92.0", "components": { "ollama-proxy": "0.26.2", - "lmstudio-proxy": "0.16.2", + "lmstudio-proxy": "0.17.0", "nvpair-node-info": "0.13.3", "nvpair-node-scanner": "0.20.3", - "nvpair-manual-nodes": "0.11.1", + "nvpair-manual-nodes": "0.12.0", "nvpair-workload-manager": "0.13.3", "nvpair-errors": "0.7.4", "nvpair-node-settings": "1.0.4", - "nvpair-ui-broker": "0.40.2", + "nvpair-ui-broker": "0.41.0", "nvpair-engine-manager": "0.17.4", "nvpair-cluster-manager": "1.1.4", "nvpair-job-scheduler": "0.4.1", - "nvpair-tui": "0.7.2" + "nvpair-tui": "0.8.0" } } From 356ddcf3abe7feeb7d8e67b7d4e48423edf4f120 Mon Sep 17 00:00:00 2001 From: Will Ford Date: Wed, 9 Sep 2026 02:47:22 +0200 Subject: [PATCH 6/7] tui: fit tables to viewport width; last column was clipped on all tabs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit bubbles' default table styles add one space of padding to each side of every cell, so rows rendered 2*n_columns wider than the width the views budget, and the table viewport's hard truncation cut the excess off the right edge — clipping the rightmost column of every tab at any terminal width (first visibly so on the Manual table's fifth column). Drop the default cell/header padding so rows render at their exact budgeted width, and add a width-sweep test asserting the last column survives. Signed-off-by: Will Ford --- services/nvpair-tui/ui/table.go | 8 ++++++- services/nvpair-tui/ui/table_test.go | 35 ++++++++++++++++++++++++++++ services/versions.json | 2 +- 3 files changed, 43 insertions(+), 2 deletions(-) create mode 100644 services/nvpair-tui/ui/table_test.go diff --git a/services/nvpair-tui/ui/table.go b/services/nvpair-tui/ui/table.go index 3d2d025c..bc15faa8 100644 --- a/services/nvpair-tui/ui/table.go +++ b/services/nvpair-tui/ui/table.go @@ -16,7 +16,13 @@ func newTable(cols []table.Column) table.Model { table.WithFocused(true), ) s := table.DefaultStyles() - s.Header = s.Header. + // The default Cell/Header styles add one space of padding to each side of + // every cell, so rows render 2*n_columns wider than the budgeted column + // widths; the table's viewport then hard-truncates rows at the terminal + // width, silently clipping the rightmost column. Keep cells at their + // exact budgeted widths instead. + s.Cell = lipgloss.NewStyle() + s.Header = lipgloss.NewStyle(). Bold(true). Foreground(colorAccent). BorderStyle(lipgloss.NormalBorder()). diff --git a/services/nvpair-tui/ui/table_test.go b/services/nvpair-tui/ui/table_test.go new file mode 100644 index 00000000..91d58cac --- /dev/null +++ b/services/nvpair-tui/ui/table_test.go @@ -0,0 +1,35 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +package ui + +import ( + "fmt" + "strings" + "testing" +) + +// TestManualTableFitsViewportWidth guards against the table body rendering +// wider than its viewport: the viewport hard-truncates every body row at the +// terminal width, so any width miscalculation silently clips the rightmost +// column. The last column's value ("yes", unique to that column here) must +// survive rendering intact at every width in the sweep (168 is a real-world +// operator width). +func TestManualTableFitsViewportWidth(t *testing.T) { + for _, w := range []int{44, 60, 80, 120, 168} { + t.Run(fmt.Sprintf("width%03d", w), func(t *testing.T) { + v := newManualView(nil) + v.SetSize(w, 20) + v.setNodes([]manualNode{{ + ID: "manual:127.0.0.1:8888", + Address: "127.0.0.1:8888", + OllamaUp: false, + NodeInfoUp: false, + OpenAIUp: true, + }}) + if !strings.Contains(v.View(), "yes") { + t.Errorf("w=%d: last column value clipped — table renders wider than the viewport", w) + } + }) + } +} diff --git a/services/versions.json b/services/versions.json index 20490cc9..bea1c950 100644 --- a/services/versions.json +++ b/services/versions.json @@ -15,6 +15,6 @@ "nvpair-engine-manager": "0.17.4", "nvpair-cluster-manager": "1.1.4", "nvpair-job-scheduler": "0.4.1", - "nvpair-tui": "0.8.0" + "nvpair-tui": "0.9.0" } } From a2d2ec5e97d080c8c0b2d3c4487c9cff34b22524 Mon Sep 17 00:00:00 2001 From: Will Ford Date: Fri, 11 Sep 2026 19:52:35 +0200 Subject: [PATCH 7/7] manual-nodes, ui-broker: stop proxy bridge churn from telemetry updates The prober change diff included msSince, which advances on every probe. That meant any manual node with node-info available emitted a state change every probe cycle, even when nothing related to routing had changed. Same-box endpoints hit this consistently. Each update re-entered the manual-to-proxy bridge and reissued the add or remove RPC. Re-adding a manual node removes and reinserts it on the proxy side, causing log churn and unnecessary priority snapshot rebuilds. msSince is only used for display. The prober store already keeps the current value for nodes/list, so remove it from change detection. Also make the bridge idempotent by skipping the RPC when the same add or remove intent has already been applied to that proxy, engine, and key. Explicit removes and prober crashes clear the record so a later add is bridged again. Respawned proxies are reseeded automatically. Signed-off-by: Will Ford --- services/nvpair-manual-nodes/manager.go | 9 ++- services/nvpair-manual-nodes/manager_test.go | 22 +++++ services/nvpair-ui-broker/broker.go | 10 +++ services/nvpair-ui-broker/manualnodes.go | 81 +++++++++++++++++-- services/nvpair-ui-broker/manualnodes_test.go | 37 +++++++++ services/versions.json | 8 +- 6 files changed, 156 insertions(+), 11 deletions(-) diff --git a/services/nvpair-manual-nodes/manager.go b/services/nvpair-manual-nodes/manager.go index d1a0510f..af02271d 100644 --- a/services/nvpair-manual-nodes/manager.go +++ b/services/nvpair-manual-nodes/manager.go @@ -380,6 +380,12 @@ func (m *Manager) probeNode(entry ManualEntry) { curFails := tn.consecutiveFails m.mu.Unlock() + // MSSince is deliberately excluded from the diff: it is the node-info + // telemetry sample's age, which advances on every probe by + // construction, so counting it as a change would emit node/updated on + // every probe cycle for every node-info-up node and make the broker + // re-bridge each one into the proxies each time. It is display-only — + // the store always holds the fresh value for nodes/list consumers. changed := prev.OllamaUp != newStatus.OllamaUp || prev.LMStudioUp != newStatus.LMStudioUp || prev.OpenAIUp != newStatus.OpenAIUp || @@ -391,8 +397,7 @@ func (m *Manager) probeNode(entry ManualEntry) { !gpusEqual(prev.GPUs, newStatus.GPUs) || !cpuEqual(prev.CPU, newStatus.CPU) || !memoryEqual(prev.Memory, newStatus.Memory) || - prev.TelemetryValid != newStatus.TelemetryValid || - prev.MSSince != newStatus.MSSince + prev.TelemetryValid != newStatus.TelemetryValid if changed { slog.Info("manual node state changed", diff --git a/services/nvpair-manual-nodes/manager_test.go b/services/nvpair-manual-nodes/manager_test.go index 543ec3d5..fa7c7178 100644 --- a/services/nvpair-manual-nodes/manager_test.go +++ b/services/nvpair-manual-nodes/manager_test.go @@ -474,6 +474,28 @@ func TestProbeNodeNoUpdateWhenStable(t *testing.T) { assertNoCaptureMethod(t, rw, "node/updated") } +// msSince is the age of the node-info telemetry sample: it advances on +// every probe by construction and is display-only (the store always holds +// the fresh value for nodes/list). It must not count as a state change, or +// every node-info-up manual node emits node/updated on every probe cycle +// and the broker re-bridges it into the proxies each time. +func TestProbeNodeNoUpdateWhenOnlyTelemetryAgeChanges(t *testing.T) { + m, rw, rt := newTestManager() + entry := ManualEntry{Name: "lab", Address: "node.local"} + m.nodes["lab"] = &trackedNode{entry: entry, status: ManualNodeStatus{ID: "lab", Address: "node.local", OllamaPort: 11434, NodeInfoPort: 14318}} + info := sampleInfo() + configureHealthyNode(rt, "node.local", []string{"llama3"}, info) + + m.probeNode(entry) + _ = readCaptureUntil(t, rw, methodIs("node/updated")) + + info.MSSince = 12347 + configureHealthyNode(rt, "node.local", []string{"llama3"}, info) + m.probeNode(entry) + + assertNoCaptureMethod(t, rw, "node/updated") +} + func TestProbeFailuresClearAvailability(t *testing.T) { m, rw, rt := newTestManager() entry := ManualEntry{Name: "lab", Address: "node.local"} diff --git a/services/nvpair-ui-broker/broker.go b/services/nvpair-ui-broker/broker.go index 8f2eb8d2..231b0820 100644 --- a/services/nvpair-ui-broker/broker.go +++ b/services/nvpair-ui-broker/broker.go @@ -306,6 +306,12 @@ type Broker struct { manualNodeKeys map[string]string manualNodeStatuses map[string]manualNodeStatusEntry + // bridgeMu guards bridgeSeen: the last intent the manual→proxy bridge + // applied per (proxy instance, engine, key) slot, so a repeated identical + // intent (telemetry-only re-probe) does not re-issue the add/remove RPC. + bridgeMu sync.Mutex + bridgeSeen map[bridgeSeenKey]bridgeIntent + // schedMu guards each engine's cached priority and generation. Per-engine // delivery locks serialize asynchronous node/set-priority calls; a stale // generation is skipped before it can overwrite a newer proxy order. @@ -383,6 +389,7 @@ func NewBroker(codec *Codec, paths workerPaths) *Broker { regCache: relay.NewRegistrationCache(), manualNodeKeys: make(map[string]string), manualNodeStatuses: make(map[string]manualNodeStatusEntry), + bridgeSeen: make(map[bridgeSeenKey]bridgeIntent), workloads: workloadstore.New(), ollamaPortReady: make(chan struct{}), lmstudioPortReady: make(chan struct{}), @@ -1396,6 +1403,9 @@ func (b *Broker) clearManualNodesState() { b.manualNodeKeys = make(map[string]string) b.manualNodeStatuses = make(map[string]manualNodeStatusEntry) b.manualMu.Unlock() + b.bridgeMu.Lock() + b.bridgeSeen = make(map[bridgeSeenKey]bridgeIntent) + b.bridgeMu.Unlock() seen := make(map[string]bool, len(keys)) for _, key := range keys { // Aliases can share a key; drop each unique claim once. diff --git a/services/nvpair-ui-broker/manualnodes.go b/services/nvpair-ui-broker/manualnodes.go index 4b35f2ec..8b54221d 100644 --- a/services/nvpair-ui-broker/manualnodes.go +++ b/services/nvpair-ui-broker/manualnodes.go @@ -7,6 +7,7 @@ import ( "context" "encoding/json" "log/slog" + "slices" "time" "nvpair-shared/noderec" @@ -175,6 +176,44 @@ type proxyManualNode struct { BasePath string `json:"base_path,omitempty"` } +// bridgeIntent is what the bridge last applied to one (proxy, engine, key) +// slot: a remove, or an add with the exact payload that was sent. +type bridgeIntent struct { + removed bool + node proxyManualNode +} + +// bridgeSeenKey identifies one bridge slot: the proxy process instance (a +// respawned proxy is a new pointer, so a restarted proxy is re-seeded +// automatically), the engine leg, and the node's operational key. +type bridgeSeenKey struct { + p *proxyProcess + engine string + key string +} + +// bridgeIntentsEqual reports whether two intents are routing-equivalent: the +// same remove, or the same add payload. Only fields the proxy payload +// carries are compared — telemetry that lives in the node's status but not +// in the payload (sample age, utilization, CPU/memory) must not force a +// re-bridge. +func bridgeIntentsEqual(a, b bridgeIntent) bool { + if a.removed != b.removed { + return false + } + if a.removed { + return true + } + n, m := a.node, b.node + return n.ID == m.ID && + n.Host == m.Host && + n.Port == m.Port && + n.BasePath == m.BasePath && + slices.Equal(n.Addresses, m.Addresses) && + slices.Equal(n.TXT, m.TXT) && + slices.Equal(n.Models, m.Models) +} + // bridgeManualNode keeps every supervised proxy's manual-node set in step with // a manual node's per-engine reachability: a node whose Ollama is up is bridged // into ollama-proxy and one whose LM Studio is up into lmstudio-proxy @@ -213,8 +252,9 @@ func (b *Broker) bridgeToProxy(p *proxyProcess, engine string, s manualNodeStatu if p == nil { return } + var intent bridgeIntent if up && s.Address != "" && port > 0 { - node := proxyManualNode{ + intent.node = proxyManualNode{ ID: key, Host: s.Address, Port: port, @@ -222,18 +262,49 @@ func (b *Broker) bridgeToProxy(p *proxyProcess, engine string, s manualNodeStatu Models: models, BasePath: basePath, } - b.callProxyManual(p, engine, "node/add-manual", node, key) + } else { + // Engine unreachable (down, or this node doesn't run it): make sure + // the proxy isn't left holding a stale manual entry it would try to + // route to. + intent.removed = true + } + + // The prober re-emits node/updated on telemetry-only wobble (sample age, + // GPU utilization), and each emission re-enters this bridge with the same + // routing intent. A repeat add-manual is not a no-op on the proxy side — + // it re-inserts the candidate (remove + add log lines, a rebuilt + // priority snapshot, and a brief window in which the node is absent) — + // so skip the RPC when the exact intent was already applied to this slot. + b.bridgeMu.Lock() + k := bridgeSeenKey{p: p, engine: engine, key: key} + if prev, ok := b.bridgeSeen[k]; ok && bridgeIntentsEqual(prev, intent) { + b.bridgeMu.Unlock() return } - // Engine unreachable (down, or this node doesn't run it): make sure the - // proxy isn't left holding a stale manual entry it would try to route to. - b.callProxyManual(p, engine, "node/remove-manual", map[string]string{"id": key}, key) + b.bridgeSeen[k] = intent + b.bridgeMu.Unlock() + + if intent.removed { + b.callProxyManual(p, engine, "node/remove-manual", map[string]string{"id": key}, key) + return + } + b.callProxyManual(p, engine, "node/add-manual", intent.node, key) } // removeManualNodeFromProxies drops a manual node from every supervised proxy. // Idempotent: a no-op for a proxy where the node was never bridged or that // isn't supervised (the proxy's RemoveManual just reports removed=false). func (b *Broker) removeManualNodeFromProxies(id string) { + // Forget the applied intents for this key: an explicit user removal (or a + // crashed prober's cleanup) must let a later re-add of the same node + // re-bridge instead of being skipped as "already applied". + b.bridgeMu.Lock() + for k := range b.bridgeSeen { + if k.key == id { + delete(b.bridgeSeen, k) + } + } + b.bridgeMu.Unlock() b.callProxyManual(b.getProxy(), "ollama", "node/remove-manual", map[string]string{"id": id}, id) b.callProxyManual(b.getLMStudioProxy(), "lmstudio", "node/remove-manual", map[string]string{"id": id}, id) // The openai leg shares lmstudio-proxy with the leg above; the duplicate diff --git a/services/nvpair-ui-broker/manualnodes_test.go b/services/nvpair-ui-broker/manualnodes_test.go index 6b898eeb..0e78df0b 100644 --- a/services/nvpair-ui-broker/manualnodes_test.go +++ b/services/nvpair-ui-broker/manualnodes_test.go @@ -162,3 +162,40 @@ func TestManualToEnrichedEndpointKeepsManualKey(t *testing.T) { t.Fatalf("address-entry storeKey = %q, want the learned uuid", got) } } + +// TestBridgeIntentEquality pins the comparison that keeps the manual→proxy +// bridge idempotent: an identical intent (same add payload, or a remove) is +// equal, while any routing-relevant difference (model list, base path, +// add↔remove) is not. The bridge skips re-issuing the RPC for an equal +// intent, so a repeated telemetry-only probe can't churn the proxy. +func TestBridgeIntentEquality(t *testing.T) { + add := bridgeIntent{node: proxyManualNode{ + ID: "k", Host: "h", Port: 8888, Addresses: []string{"h"}, + Models: []string{"m1"}, BasePath: "/v1", + }} + rem := bridgeIntent{removed: true} + + if !bridgeIntentsEqual(add, bridgeIntent{node: proxyManualNode{ + ID: "k", Host: "h", Port: 8888, Addresses: []string{"h"}, + Models: []string{"m1"}, BasePath: "/v1", + }}) { + t.Fatal("identical add intents must be equal") + } + if !bridgeIntentsEqual(rem, bridgeIntent{removed: true}) { + t.Fatal("identical remove intents must be equal") + } + + changedModels := add + changedModels.node.Models = []string{"m1", "m2"} + if bridgeIntentsEqual(add, changedModels) { + t.Fatal("a model-list change must not be equal") + } + changedPath := add + changedPath.node.BasePath = "" + if bridgeIntentsEqual(add, changedPath) { + t.Fatal("a base-path change must not be equal") + } + if bridgeIntentsEqual(add, rem) { + t.Fatal("add and remove intents must not be equal") + } +} diff --git a/services/versions.json b/services/versions.json index bea1c950..88fcd043 100644 --- a/services/versions.json +++ b/services/versions.json @@ -1,17 +1,17 @@ { "$comment": "Single source of truth for all version numbers. See VERSIONING.md for bump rules.", - "product": "0.92.0", - "installer": "0.92.0", + "product": "0.93.0", + "installer": "0.93.0", "components": { "ollama-proxy": "0.26.2", "lmstudio-proxy": "0.17.0", "nvpair-node-info": "0.13.3", "nvpair-node-scanner": "0.20.3", - "nvpair-manual-nodes": "0.12.0", + "nvpair-manual-nodes": "0.13.0", "nvpair-workload-manager": "0.13.3", "nvpair-errors": "0.7.4", "nvpair-node-settings": "1.0.4", - "nvpair-ui-broker": "0.41.0", + "nvpair-ui-broker": "0.42.0", "nvpair-engine-manager": "0.17.4", "nvpair-cluster-manager": "1.1.4", "nvpair-job-scheduler": "0.4.1",