Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 5 additions & 1 deletion docs/architecture.mdx
Original file line number Diff line number Diff line change
Expand Up @@ -816,7 +816,11 @@ memberships and retire removed or re-addressed senders. The responder reuses its
senders for announcements and replies too. Shutdown closes the sockets after the
responder's goodbye. A standalone `Browser.Poll` owns a one-shot transport; use
`Run` for periodic discovery. The scan cadence and per-interface send-failure
feedback used by address selection are unchanged.
feedback used by address selection are unchanged. Partial browser receive-setup
failures retry only the failed family binds or interface memberships at scan
boundaries, with exponential backoff from 1 second to a 60-second cap. Successful
sockets and memberships stay in place; a changed network snapshot rebuilds setup
immediately rather than waiting for that backoff.

This reduces measured socket and interface-query churn. It does **not** establish
that discovery caused the macOS kernel panics or WLAN failures reported in
Expand Down
29 changes: 28 additions & 1 deletion services/nvpair-engine-manager/MANIFEST.md
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,34 @@ SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All
SPDX-License-Identifier: Apache-2.0
-->

# Editable launch manifest bindings
# Engine manifest capabilities

## Command-mode shutdown confirmation

`runtime.stop.wait_for_exit` is an optional boolean (default `false`). It
requires `runtime.mode: "command"`, a nonempty `runtime.stop.cmd`, and a
`runtime.ready` probe; manifests enabling it without these are rejected.

By default, command-mode shutdown confirms three consecutive connection
refusals on the readiness listener. A persistent host process, such as LM
Studio after `lms server stop`, need not exit; its server can be restarted.
Set `wait_for_exit: true` only when the app must fully exit before restarting.
Only bundled oMLX enables it:

```json
"stop": { "cmd": ["{cli}", "stop"], "wait_for_exit": true }
```

With this capability, PAIR also waits for the serving PID observed before the
stop command to exit, when the host can resolve it. If PID lookup is unavailable,
listener closure remains the available evidence. The stop command is bounded by
`max(grace_s, 5s)` and confirmation by `grace_s` (default 5s), without extending
the deadline. Timeout retains running state and reports an error. The state
mutex is never held across lookup, command execution, or confirmation; the
lifecycle operation lock still serializes stop/start. Failed-start cleanup uses
the same confirmation policy.

## Editable launch manifest bindings

The engine-manager runner remains manifest-driven. `runtime.editable_launch`
identifies the fixed startup arguments and PAIR-owned networking controls; it
Expand Down
8 changes: 5 additions & 3 deletions services/nvpair-engine-manager/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -600,9 +600,11 @@ terminate is the only signal that actually stops them.

Command-mode stop (including oMLX) gives the official stop CLI
`max(grace_s, 5s)` to return, then up to `grace_s` (default 5s) for shutdown.
It confirms three consecutive connection refusals **and**, when the host can
resolve it, exit of the process observed on the port before stop. Closing the
port alone does not mean app teardown is complete. The state mutex is never
It confirms three consecutive connection refusals. Only manifests opting into
`runtime.stop.wait_for_exit` (bundled oMLX) also require exit of the process
observed on the port before stop, when the host can resolve it. LM Studio stops
only its HTTP server; its persistent host need not exit for stop/restart.
See [MANIFEST.md](MANIFEST.md#command-mode-shutdown-confirmation). The state mutex is never
held across these waits; lifecycle operations remain serialized so an immediate
start cannot overtake stop. A timeout leaves the engine running, not falsely
stopped, and is returned to the caller.
Expand Down
13 changes: 8 additions & 5 deletions services/nvpair-engine-manager/lifecycle.go
Original file line number Diff line number Diff line change
Expand Up @@ -561,12 +561,14 @@ func (e *Executor) runCommandStop(st *engineState, engine string, rt Runtime, po
// Bound the stop CLI (e.g. `lms server stop`): the broker no longer
// force-kills engine-manager on a timeout, so an unbounded stop command
// would wedge StopAll and, in turn, the whole app shutdown.
// Snapshot the serving PID before the CLI can close its listener. The
// daemon can remain alive tearing down state after the port is released.
// Only engines declaring wait_for_exit need the serving PID to exit:
// some apps tear down after releasing the port, while others (LM Studio)
// intentionally keep their host process alive after stopping the server.
// Snapshot the PID before the CLI can close its listener.
// No state mutex is held during lookup, command execution, or this wait;
// opMu still serializes lifecycle operations, as it does for readiness.
pid := 0
if rt.Ready != nil {
if sp.WaitForExit && rt.Ready != nil {
pid, _, _ = pidOnPort(port)
// Never wait for this service itself to exit (e.g. a misconfigured
// probe targeting engine-manager). Only an independent daemon counts.
Expand Down Expand Up @@ -648,8 +650,9 @@ func (e *Executor) waitUnavailableContext(parent context.Context, p *Probe, port
return e.waitDownContext(parent, p, port, 0, timeout)
}

// waitCommandUnavailable requires both transport closure and exit of the PID
// observed before stop. If the host cannot resolve a PID, transport remains
// waitCommandUnavailable requires transport closure and, for engines opting
// into wait_for_exit, exit of the PID observed before stop. Otherwise pid is 0.
// If the host cannot resolve a PID, transport remains
// the best available evidence; the deadline is never extended by polling.
func (e *Executor) waitCommandUnavailable(p *Probe, port, pid int, timeout time.Duration) bool {
return e.waitDownContext(context.Background(), p, port, pid, timeout)
Expand Down
2 changes: 1 addition & 1 deletion services/nvpair-engine-manager/manifests/omlx.json
Original file line number Diff line number Diff line change
Expand Up @@ -50,7 +50,7 @@
"port": 1235,
"bind": "127.0.0.1",
"start": [["{cli}", "start"]],
"stop": { "cmd": ["{cli}", "stop"] },
"stop": { "cmd": ["{cli}", "stop"], "wait_for_exit": true },
"ready": { "http": "http://127.0.0.1:{port}/health", "status": 200, "timeout_s": 45 },
"health": { "http": "http://127.0.0.1:{port}/health", "status": 200, "interval_s": 10 }
},
Expand Down
8 changes: 8 additions & 0 deletions services/nvpair-engine-manager/registry.go
Original file line number Diff line number Diff line change
Expand Up @@ -238,6 +238,9 @@ type StopSpec struct {
Signal string `json:"signal,omitempty"` // "term" (default) | "kill"
Cmd []string `json:"cmd,omitempty"` // optional explicit stop command
GraceS int `json:"grace_s,omitempty"`
// WaitForExit requires command-mode shutdown to confirm the serving PID
// exits as well as releasing the listener. Most apps only stop the server.
WaitForExit bool `json:"wait_for_exit,omitempty"`
}

// Action is a config-declared operation exposed over engine:action.
Expand Down Expand Up @@ -778,6 +781,11 @@ func (p *Platform) validate(key string) error {
}
}
}
if sp := p.Runtime.Stop; sp != nil && sp.WaitForExit {
if p.Runtime.modeOrDefault() != "command" || len(sp.Cmd) == 0 || p.Runtime.Ready == nil {
return fmt.Errorf("platform %q: runtime.stop.wait_for_exit requires command mode, stop.cmd, and runtime.ready", key)
}
}
if err := validateProbe(key, "ready", p.Runtime.Ready); err != nil {
return err
}
Expand Down
54 changes: 54 additions & 0 deletions services/nvpair-engine-manager/registry_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@ package main

import (
"encoding/json"
"fmt"
"os"
"path/filepath"
"slices"
Expand Down Expand Up @@ -69,6 +70,40 @@ func TestValidateAcceptsCommandModeAndCmdAction(t *testing.T) {
}
}

func TestValidateStopWaitForExit(t *testing.T) {
for _, tc := range []struct {
name, mode, wait, cmd, ready string
wantErr bool
}{
{"enabled command", "command", "true", `["cli","stop"]`, `{"tcp":"127.0.0.1:{port}"}`, false},
{"disabled command", "command", "false", `["cli","stop"]`, `null`, false},
{"disabled process", "process", "false", `[]`, `null`, false},
{"process mode", "process", "true", `["cli","stop"]`, `{"tcp":"127.0.0.1:{port}"}`, true},
{"missing stop command", "command", "true", `[]`, `{"tcp":"127.0.0.1:{port}"}`, true},
{"missing readiness probe", "command", "true", `["cli","stop"]`, `null`, true},
{"string instead of boolean", "command", `"true"`, `["cli","stop"]`, `{"tcp":"127.0.0.1:{port}"}`, true},
{"number instead of boolean", "command", "1", `["cli","stop"]`, `{"tcp":"127.0.0.1:{port}"}`, true},
} {
t.Run(tc.name, func(t *testing.T) {
data := fmt.Sprintf(`{
"engine":"test", "display_name":"Test", "manifest_version":1,
"platforms":{"linux/amd64":{"runtime":{
"mode":%q, "bin":"cli", "start":[["cli","start"]],
"stop":{"cmd":%s,"wait_for_exit":%s}, "ready":%s
}}}
}`, tc.mode, tc.cmd, tc.wait, tc.ready)
var m Manifest
err := json.Unmarshal([]byte(data), &m)
if err == nil {
err = m.Validate()
}
if (err != nil) != tc.wantErr {
t.Fatalf("manifest validation error = %v, wantErr=%v", err, tc.wantErr)
}
})
}
}

func TestValidateAcceptsUnpinnedFetch(t *testing.T) {
m := validManifest()
p := m.Platforms["linux/amd64"]
Expand Down Expand Up @@ -384,6 +419,25 @@ func TestBundledManifestsMerge(t *testing.T) {
}
}

func TestBundledStopWaitForExit(t *testing.T) {
reg := NewRegistry()
if err := reg.LoadFS(bundledManifests, "manifests"); err != nil {
t.Fatal(err)
}
for _, engine := range []string{"omlx", "lmstudio", "ollama"} {
m, ok := reg.Get(engine)
if !ok {
t.Fatalf("missing bundled engine %s", engine)
}
for key, p := range m.Platforms {
wait := p.Runtime.Stop != nil && p.Runtime.Stop.WaitForExit
if wait != (engine == "omlx") {
t.Errorf("%s %s: wait_for_exit=%v, only oMLX requires app exit before restart", engine, key, wait)
}
}
}
}

// TestBundledOllamaReadinessBudget pins the finite startup allowance used by
// every supported platform. Ollama does not serve /api/version until GPU
// discovery completes, which can exceed the previous 30-second allowance.
Expand Down
60 changes: 59 additions & 1 deletion services/nvpair-engine-manager/shutdown_tail_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
package main

import (
"context"
"errors"
"os/exec"
"path/filepath"
Expand All @@ -12,6 +13,63 @@ import (
"time"
)

func TestCommandStopAndRestartWithPersistentHost(t *testing.T) {
marker := filepath.Join(t.TempDir(), "server")
port, err := freePort()
if err != nil {
t.Fatal(err)
}
cmd := exec.Command(fakeEngineForTest(t), "persistent-host", marker, "127.0.0.1:"+strconv.Itoa(port))
configureSysProcAttr(cmd)
if err := cmd.Start(); err != nil {
t.Fatal(err)
}
done := make(chan struct{})
go func() { _ = cmd.Wait(); close(done) }()
t.Cleanup(func() { killTestProcessGroup(cmd.Process.Pid); <-done })
waitPortServing(t, port)
if pid, _, ok := pidOnPort(port); !ok || pid != cmd.Process.Pid {
t.Skip("host cannot resolve listener PID")
}
m := testEngineManifest(fakeEngineForTest(t))
p := m.Platforms[hostKey()]
p.Runtime.Mode = "command"
p.Runtime.Port = port
p.Runtime.Start = [][]string{{fakeEngineForTest(t), "touch", marker + "-start"}}
p.Runtime.Stop = &StopSpec{Cmd: []string{fakeEngineForTest(t), "touch", marker + "-stop"}, GraceS: 1}
p.Runtime.Health = nil
m.Platforms[hostKey()] = p
ex := newTestExecutor(t, m)
st, _ := ex.state("fake")
st.running, st.port = true, port

if err := ex.Stop("fake"); err != nil {
t.Fatalf("stop must succeed when the server closes but its host stays alive: %v", err)
}
if status, _ := ex.Status("fake"); status.Running || status.Healthy || portServing(port) {
t.Fatalf("stopped server must be reported down: %+v", status)
}
select {
case <-done:
t.Fatal("stop terminated the persistent host")
default:
}
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
if err := ex.Start(ctx, "fake"); err != nil {
t.Fatalf("start after stop: %v", err)
}
if err := ex.Restart(ctx, "fake"); err != nil {
t.Fatalf("restart with a persistent host: %v", err)
}
if status, _ := ex.Status("fake"); !status.Running || !status.Healthy || !portServing(port) {
t.Fatalf("restarted server must be healthy: %+v", status)
}
if pid, _, ok := pidOnPort(port); !ok || pid != cmd.Process.Pid {
t.Fatalf("restart must reuse the persistent host, got pid=%d resolved=%v", pid, ok)
}
}

func TestCommandStopWaitsForProcessExitAfterPortCloses(t *testing.T) {
testCommandStopTail(t, "900ms", 2, false)
}
Expand Down Expand Up @@ -43,7 +101,7 @@ func testCommandStopTail(t *testing.T, delay string, grace int, wantTimeout bool
p := m.Platforms[hostKey()]
p.Runtime.Mode = "command"
p.Runtime.Port = port
p.Runtime.Stop = &StopSpec{Cmd: []string{fakeEngineForTest(t), "touch", marker}, GraceS: grace}
p.Runtime.Stop = &StopSpec{Cmd: []string{fakeEngineForTest(t), "touch", marker}, GraceS: grace, WaitForExit: true}
p.Runtime.Health = nil
m.Platforms[hostKey()] = p
ex := newTestExecutor(t, m)
Expand Down
6 changes: 3 additions & 3 deletions services/nvpair-engine-manager/shutdown_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -142,6 +142,8 @@ func testStopAllDetachedCommandDaemon(t *testing.T, command string) {
platform.Runtime.Health = nil
manifest.Platforms[hostKey()] = platform
ex := newTestExecutor(t, manifest)
// This must run even if fixture setup/readiness fails before explicit StopAll.
t.Cleanup(ex.StopAll)

// Model a control CLI that launches a detached daemon and returns before
// that daemon becomes ready. The daemon deliberately answers 503 until its
Expand Down Expand Up @@ -195,9 +197,7 @@ func testStopAllDetachedCommandDaemon(t *testing.T, command string) {
if err := <-daemonStarted; err != nil {
t.Fatalf("detached command daemon: %v", err)
}
if !portServing(port) {
t.Fatal("detached command daemon did not start serving")
}
waitPortServing(t, port)
if command == "touch" {
// A marker written by the CLI does not prove cmd.Wait has returned.
// Observe readiness I/O to synchronize with the intended startup phase.
Expand Down
38 changes: 32 additions & 6 deletions services/nvpair-engine-manager/status_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -254,15 +254,41 @@ func TestStartDoesNotWaitForPassiveProbe(t *testing.T) {
awaitProbeStart(t, fixture.probeStarted)
port, err := freePort()
require.NoError(t, err)
// Keep the operation observable even if startup finishes between polls.
// Persistence is the last step of StartWith, after spawning and readiness.
ex.desired.mu.Lock()
var persistOnce sync.Once
allowPersistence := func() { persistOnce.Do(ex.desired.mu.Unlock) }
t.Cleanup(func() {
allowPersistence()
fixture.releaseProbe()
ex.StopAll()
})
started := make(chan error, 1)
go func() { started <- ex.StartWith(context.Background(), "fake", startOpts{Port: port}) }()
defer func() { fixture.releaseProbe(); _ = ex.Stop("fake") }()
require.NoError(t, awaitResult(t, started))
// Only lifecycle entry has this short deadline, not real process startup.
require.Eventually(t, func() bool {
st.mu.Lock()
defer st.mu.Unlock()
return st.opDone != nil
}, 2*time.Second, time.Millisecond, "Start waited for a passive readiness probe")
select {
case <-finished:
t.Fatal("passive probe finished before lifecycle entry was observed")
default:
}
fixture.releaseProbe()
got := awaitResult(t, finished)
require.NoError(t, got.err)
require.True(t, got.status.Running)
require.Equal(t, port, got.status.Port)
require.NoError(t, awaitResult(t, finished).err)
allowPersistence()
select {
case err := <-started:
require.NoError(t, err)
case <-time.After(15 * time.Second):
t.Fatal("engine startup did not finish after lifecycle entry")
}
status := ex.snapshot("fake", st)
require.True(t, status.Running)
require.Equal(t, port, status.Port)
}

type statusResult struct {
Expand Down
30 changes: 30 additions & 0 deletions services/nvpair-engine-manager/testdata/fakeengine/main.go
Original file line number Diff line number Diff line change
Expand Up @@ -206,6 +206,36 @@ func main() {
os.Exit(1)
}
return
case "persistent-host":
// Model an app whose HTTP server can stop/restart without exiting.
var server *http.Server
start := func() {
ln, err := net.Listen("tcp", os.Args[3])
if err != nil {
os.Exit(2)
}
server = &http.Server{Handler: http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
w.WriteHeader(http.StatusOK)
})}
go func(s *http.Server) { _ = s.Serve(ln) }(server)
}
start()
for {
if _, err := os.Stat(os.Args[2] + "-stop"); err == nil {
if server != nil {
_ = server.Close()
server = nil
}
_ = os.Remove(os.Args[2] + "-stop")
}
if _, err := os.Stat(os.Args[2] + "-start"); err == nil {
if server == nil {
start()
}
_ = os.Remove(os.Args[2] + "-start")
}
time.Sleep(10 * time.Millisecond)
}
case "shutdown-tail":
// Close the listener before asynchronous process teardown completes.
ln, err := net.Listen("tcp", os.Args[3])
Expand Down
5 changes: 5 additions & 0 deletions services/nvpair-node-scanner/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,11 @@ on interface-change notifications rather than re-enumerated on every five-second
scan. The responder reuses per-interface senders for announcements and replies;
removed or re-addressed interfaces retire their cached senders, including when
all eligible interfaces disappear. Shutdown sends the goodbye before closing them.
If only part of browser receive setup succeeds, failed address-family binds and
interface memberships retry at scan boundaries with exponential backoff (1 second
up to 60 seconds), without replacing working sockets or rejoining working
memberships. A network snapshot change rebuilds setup immediately; a failed
established receiver is rebuilt on the next scan.

Every scan still performs real per-interface writes, preserving the consecutive
send-failure evidence that address selection consumes. Source binding preserves
Expand Down
Loading
Loading