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/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/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) + } + } +} 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/nvpair-manual-nodes/manager.go b/services/nvpair-manual-nodes/manager.go index 55a4040a..af02271d 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] @@ -329,17 +380,24 @@ 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 || 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) || - prev.TelemetryValid != newStatus.TelemetryValid || - prev.MSSince != newStatus.MSSince + prev.TelemetryValid != newStatus.TelemetryValid if changed { slog.Info("manual node state changed", @@ -403,23 +461,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 +544,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 +554,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 +658,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 +771,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 +806,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 +834,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..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"} @@ -689,3 +711,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) + } +} 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) + } + } +} 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/nvpair-ui-broker/broker.go b/services/nvpair-ui-broker/broker.go index 0d189578..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{}), @@ -1256,11 +1263,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) @@ -1357,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. @@ -3159,7 +3208,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..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" @@ -28,7 +29,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 +87,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 +107,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 +119,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 +131,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 +170,48 @@ 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"` +} + +// 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 @@ -160,42 +227,90 @@ 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 } + var intent bridgeIntent if up && s.Address != "" && port > 0 { - node := proxyManualNode{ + intent.node = proxyManualNode{ ID: key, Host: s.Address, Port: port, Addresses: []string{s.Address}, 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 + } + b.bridgeSeen[k] = intent + b.bridgeMu.Unlock() + + if intent.removed { + b.callProxyManual(p, engine, "node/remove-manual", map[string]string{"id": key}, key) 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.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 + // 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..0e78df0b --- /dev/null +++ b/services/nvpair-ui-broker/manualnodes_test.go @@ -0,0 +1,201 @@ +// 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) + } +} + +// 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/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 diff --git a/services/versions.json b/services/versions.json index 29d8c230..88fcd043 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.93.0", + "installer": "0.93.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.13.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.42.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.9.0" } }