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
49 changes: 41 additions & 8 deletions services/nvpair-engine-manager/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -156,8 +156,8 @@ This does not extend the supported platform list.
Each fake-engine executable (including test copies) has a per-test owner under
the run's `nvpair-em-test-*` directory. Before doing any work, the fake records
its PID and executable identity there; detached children inherit registration.
A `t.Cleanup` sweep force-kills owned process groups on Unix (process trees on
Windows) and fails the test if any remain. `TestMain` also checks registered
A `t.Cleanup` sweep force-kills owned process groups on Unix (individually
pinned, identity-checked processes on Windows) and fails the test if any remain. `TestMain` also checks registered
children and processes whose executables are under this run's directory, sweeps
survivors, and fails the run if a per-test cleanup was missed. Sweeps verify
executable identity before signaling PIDs and never target other test runs.
Expand Down Expand Up @@ -580,7 +580,7 @@ numbers and never add the two together: they count overlapping work.
```
NotInstalled --engine:install--> (HTTPS download + verify-if-pinned + user-mode run) --> Stopped
Stopped --engine:start----> (adopt if already serving the port, else spawn) --> Running --health--> Running
Running --engine:stop-----> (stop signal, wait for exit; no timeout) --> Stopped
Running --engine:stop-----> (platform stop / bounded official stop CLI) --> Stopped
```

Detect uses the manifest's executable candidates (`runtime.cli` when explicitly
Expand All @@ -592,11 +592,10 @@ unexpected exit is reported. The bundled Ollama manifest allows up to ten
minutes for startup because GPU discovery can exceed the previous 30-second
allowance on supported Windows systems. The deadline remains finite: if Ollama
never serves its readiness endpoint, engine-manager stops the owned process and
reports the failed start. Stop sends one stop signal and waits for the engine
to exit, with no timeout: SIGTERM to the process group on Unix (graceful, no
SIGKILL escalation), and `taskkill /T /F` on Windows — where the windowless
engines we spawn can't receive a graceful (non-`/F`) close, so a forced
terminate is the only signal that actually stops them.
reports the failed start. Process-mode Stop preserves SIGTERM to the process
group on Unix (graceful, no SIGKILL escalation or exit timeout). On Windows it
terminates the owned job, closes its kill-on-close handle, and bounds the final
exit wait to five seconds. There is no PID-addressed tree kill.

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.
Expand All @@ -609,6 +608,40 @@ 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.

### Windows process ownership

One shared kill-on-close job helper is used by downloads and all Windows
processes spawned by engine-manager:

| Launch | Job membership and lifetime |
| --- | --- |
| Process-mode engine server | Gated hidden-console launcher, server, and all descendants; no breakaway. Stop terminates the job; launcher exit also closes the job and kills leftovers. |
| Model download | Same launcher and full descendant job, preserving the existing bounded console-interrupt/retry/kill sequence. |
| Action CLIs, install/uninstall helpers, console-interrupt helper | One job per invocation, including launcher and descendants; no breakaway. Context cancellation terminates the job; completion closes it. Commands have a 15-minute safety ceiling in addition to caller deadlines, with one-second inherited-output drainage. |
| Command-mode start/stop CLIs (`lms server start/stop`, `omlx stop` where supported) | The actual CLI starts suspended and is assigned before its initial thread resumes. Its job owns **only the control CLI**; silent child breakaway deliberately preserves the command-mode contract that the vendor owns its daemon tree. These children are not reclaimed by job termination or by PID. The existing bounded official stop command and shutdown confirmation remain authoritative. |

Only command-mode lifecycle controls get that child-breakaway exception; normal
CLIs and install/uninstall helpers do not. All jobs and process handles remain
non-inheritable. The gated launcher inherits only its start event, which it
closes before launching the target. Assignment/resume failures terminate the
owned process and close the job; cancellation cannot resume an unassigned or
already-cancelled launch. No state mutex or job-handle mutex covers a process
wait. Existing lifecycle-operation serialization remains unchanged.

No `taskkill` calls remain. Legacy adopted Windows orphans have no job: reclaim
opens a query/terminate handle when identifying the listener and records its
creation time and image. That handle stays open through termination; immediately
before terminating, reclaim checks the creation time and image through the same
handle, and re-reads the TCP listener table to confirm it still owns the managed
port. Any failed check declines termination. The pinned handle prevents PID reuse
between checks. Reclaim terminates **that handle only**, then closes it exactly
once on every path. It does not enumerate or kill descendants for which ownership
cannot be proved. Foreign listeners and
vendor-managed daemon trees remain untouched.

Windows runtime behavior is untested here; cross-compilation and injected-job
unit tests do not establish runtime behavior on Windows.

### Adoption — start may attach to an engine it didn't launch

Before spawning, `engine:start` **probes the chosen port's readiness
Expand Down
2 changes: 1 addition & 1 deletion services/nvpair-engine-manager/actions.go
Original file line number Diff line number Diff line change
Expand Up @@ -321,7 +321,7 @@ func (e *Executor) runCommandOutput(ctx context.Context, argv []string) (string,
}
cmd := exec.CommandContext(ctx, argv[0], argv[1:]...)
configureSysProcAttr(cmd)
out, err := cmd.Output()
out, err := ownedOutput(cmd)
if err != nil {
if ee, ok := err.(*exec.ExitError); ok {
return "", fmt.Errorf("%v: %s", err, strings.TrimSpace(string(ee.Stderr)))
Expand Down
18 changes: 18 additions & 0 deletions services/nvpair-engine-manager/command_context.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,18 @@
// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
// SPDX-License-Identifier: Apache-2.0

package main

import "context"

type controlCommandKey struct{}

// Only command-mode start/stop declares an independently managed daemon tree.
// Install/uninstall helpers and action CLIs retain normal tree ownership.
func withControlCommand(ctx context.Context) context.Context {
return context.WithValue(ctx, controlCommandKey{}, true)
}
func isControlCommand(ctx context.Context) bool {
control, _ := ctx.Value(controlCommandKey{}).(bool)
return control
}
30 changes: 30 additions & 0 deletions services/nvpair-engine-manager/command_process.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,30 @@
// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
// SPDX-License-Identifier: Apache-2.0

//go:build windows

package main

import (
"bytes"
"errors"
"os/exec"
)

func ownedCombinedOutput(cmd *exec.Cmd, detachedControl bool) ([]byte, error) {
var output bytes.Buffer
cmd.Stdout, cmd.Stderr = &output, &output
err := runOwnedCommandMode(cmd, detachedControl)
return output.Bytes(), err
}

func ownedOutput(cmd *exec.Cmd) ([]byte, error) {
var output, diagnostic bytes.Buffer
cmd.Stdout, cmd.Stderr = &output, &diagnostic
err := runOwnedCommand(cmd)
var exit *exec.ExitError
if errors.As(err, &exit) {
exit.Stderr = diagnostic.Bytes()
}
return output.Bytes(), err
}
13 changes: 13 additions & 0 deletions services/nvpair-engine-manager/command_process_unix.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,13 @@
// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
// SPDX-License-Identifier: Apache-2.0

//go:build !windows

package main

import "os/exec"

func ownedCombinedOutput(cmd *exec.Cmd, _ bool) ([]byte, error) { return cmd.CombinedOutput() }
func ownedOutput(cmd *exec.Cmd) ([]byte, error) { return cmd.Output() }
func runOwnedCommandMode(cmd *exec.Cmd, _ bool) error { return cmd.Run() }
func runOwnedCommand(cmd *exec.Cmd) error { return cmd.Run() }
25 changes: 25 additions & 0 deletions services/nvpair-engine-manager/command_process_windows.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,25 @@
// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
// SPDX-License-Identifier: Apache-2.0

//go:build windows

package main

import (
"os/exec"
"time"
)

// Even a caller without a context deadline cannot wait forever for a helper.
// More restrictive caller deadlines still cancel the entire owned job.
const windowsCommandLimit = 15 * time.Minute

func runOwnedCommand(cmd *exec.Cmd) error { return runOwnedCommandMode(cmd, false) }

func runOwnedCommandMode(cmd *exec.Cmd, detachedControl bool) error {
owner, err := configureProcessJob(cmd, detachedControl)
if err != nil {
return err
}
return runJobCommand(cmd, owner, windowsCommandLimit)
}
Original file line number Diff line number Diff line change
Expand Up @@ -28,8 +28,9 @@ func TestLMSCompletionStopsDownloaderChildAfterLauncherExit(t *testing.T) {
default:
}
}}
bin := fakeEngineForTest(t)
go func() {
_, err := runLMSDownloadCommand(ctx, []string{fakeEngineForTest(t), "shutdown-download-fixture", marker, "exit-parent", childMarker}, output)
_, err := runLMSDownloadCommand(ctx, []string{bin, "shutdown-download-fixture", marker, "exit-parent", childMarker}, output)
done <- err
}()
select {
Expand All @@ -53,11 +54,7 @@ func TestLMSCompletionStopsDownloaderChildAfterLauncherExit(t *testing.T) {
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() {
if child, err := os.FindProcess(childPID); err == nil {
_ = child.Kill()
}
})
t.Cleanup(func() { killOwnedTestPID(childPID, bin) })
address, err := os.ReadFile(childMarker)
if err != nil {
t.Fatal(err)
Expand Down
13 changes: 4 additions & 9 deletions services/nvpair-engine-manager/download_shutdown_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -53,6 +53,7 @@ func testRunStopsDownloader(t *testing.T, ignore, midCancel, tree, exitEarly, us
t.Fatal(err)
}
}
bin := fakeEngineForTest(t)
marker := filepath.Join(t.TempDir(), "downloader-pid")
childMarker := marker + "-child"
events, unsub := ex.progress.subscribe("lmstudio")
Expand All @@ -77,7 +78,7 @@ func testRunStopsDownloader(t *testing.T, ignore, midCancel, tree, exitEarly, us
go func() {
defer close(pullDone)
_, _ = ex.trackedPull(ctx, "lmstudio", "fixture", "shutdown-pull", true, func(ctx context.Context) (json.RawMessage, error) {
argv := []string{fakeEngineForTest(t), "shutdown-download-fixture", marker, mode}
argv := []string{bin, "shutdown-download-fixture", marker, mode}
if tree {
argv = append(argv, childMarker)
}
Expand All @@ -99,9 +100,7 @@ func testRunStopsDownloader(t *testing.T, ignore, midCancel, tree, exitEarly, us
t.Fatal(err)
}
t.Cleanup(func() {
if proc, err := os.FindProcess(pid); err == nil {
_ = proc.Kill()
}
killOwnedTestPID(pid, bin)
cancel()
select {
case <-pullDone:
Expand Down Expand Up @@ -130,11 +129,7 @@ func testRunStopsDownloader(t *testing.T, ignore, midCancel, tree, exitEarly, us
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() {
if proc, err := os.FindProcess(childPID); err == nil {
_ = proc.Kill()
}
})
t.Cleanup(func() { killOwnedTestPID(childPID, bin) })
conn, err := net.DialTimeout("tcp", childAddress, time.Second)
if err != nil {
t.Fatal("child not alive before shutdown:", err)
Expand Down
62 changes: 12 additions & 50 deletions services/nvpair-engine-manager/download_windows.go
Original file line number Diff line number Diff line change
Expand Up @@ -20,54 +20,24 @@ import (
)

type downloadProcess struct {
cmd *exec.Cmd
job, gate windows.Handle
cmd *exec.Cmd
owner *processJob
job windows.Handle
}

func (p *downloadProcess) close() {
_ = windows.CloseHandle(p.gate)
_ = windows.CloseHandle(p.job) // KILL_ON_JOB_CLOSE also covers worker crashes.
}
func (p *downloadProcess) close() { p.owner.release() }

func configureDownloadProcess(cmd *exec.Cmd) (*downloadProcess, error) {
executable, err := os.Executable()
if err != nil {
return nil, err
}
job, err := windows.CreateJobObject(nil, nil)
if err != nil {
return nil, err
}
p := &downloadProcess{cmd: cmd, job: job}
limits := windows.JOBOBJECT_EXTENDED_LIMIT_INFORMATION{}
limits.BasicLimitInformation.LimitFlags = windows.JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE
if _, err := windows.SetInformationJobObject(job, windows.JobObjectExtendedLimitInformation, uintptr(unsafe.Pointer(&limits)), uint32(unsafe.Sizeof(limits))); err != nil {
p.close()
return nil, err
}
p.gate, err = windows.CreateEvent(nil, 1, 0, nil)
owner, err := configureProcessJob(cmd, false)
if err != nil {
p.close()
return nil, err
}
handle := p.gate
if err := windows.SetHandleInformation(handle, windows.HANDLE_FLAG_INHERIT, windows.HANDLE_FLAG_INHERIT); err != nil {
p.close()
return nil, err
}
// Keep the private-console launcher, but gate CLI creation until the
// launcher belongs to its job: no child can escape an assignment race.
cmd.Args = append([]string{executable, "--run-download", strconv.FormatUint(uint64(handle), 10), cmd.Path}, cmd.Args[1:]...)
cmd.Path = executable
cmd.SysProcAttr = &syscall.SysProcAttr{
HideWindow: true, CreationFlags: windows.CREATE_NEW_CONSOLE,
AdditionalInheritedHandles: []syscall.Handle{syscall.Handle(handle)},
}
return p, nil
native := owner.ops.(*windowsProcessJob)
return &downloadProcess{cmd: cmd, owner: owner, job: native.job}, nil
}

func killDownload(_ context.Context, process *downloadProcess) error {
return windows.TerminateJobObject(process.job, 1)
return process.owner.stop()
}
func terminateDownload(ctx context.Context, process *downloadProcess) error {
return interruptDownload(ctx, process)
Expand Down Expand Up @@ -106,19 +76,11 @@ func waitDownloadGroup(ctx context.Context, process *downloadProcess) error {
// addresses that PID, whereas forced cleanup addresses the job. Assign the
// job before opening the start gate and before cmd.Wait can reap the launcher.
func pinDownloadProcess(process *downloadProcess) (func(), error) {
handle, err := windows.OpenProcess(windows.SYNCHRONIZE|windows.PROCESS_QUERY_LIMITED_INFORMATION|windows.PROCESS_SET_QUOTA|windows.PROCESS_TERMINATE, false, uint32(process.cmd.Process.Pid))
if err != nil {
return nil, err
}
if err := windows.AssignProcessToJobObject(process.job, handle); err != nil {
_ = windows.CloseHandle(handle)
return nil, err
}
if err := windows.SetEvent(process.gate); err != nil {
_ = windows.CloseHandle(handle)
if err := process.owner.activate(); err != nil {
return nil, err
}
return func() { _ = windows.CloseHandle(handle) }, nil
// The shared owner pins the launcher until close, including pipe drainage.
return func() {}, nil
}

// runDownloadProcess owns the CLI in a private hidden console. Reset only this
Expand Down Expand Up @@ -203,7 +165,7 @@ func interruptDownload(parent context.Context, process *downloadProcess) error {
defer cancel()
helper := exec.CommandContext(ctx, executable, "--interrupt-download", strconv.Itoa(cmd.Process.Pid))
configureSysProcAttr(helper)
return helper.Run()
return runOwnedCommand(helper)
}

// Download launch and console attachment run in helpers so engine-manager's
Expand Down
9 changes: 3 additions & 6 deletions services/nvpair-engine-manager/e2e_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -133,7 +133,8 @@ func TestE2EPrepareShutdownCancelsStartingEngine(t *testing.T) {
cfg := t.TempDir()
home := t.TempDir()
pidFile := filepath.Join(t.TempDir(), "fake.pid")
manifest := testEngineManifest(fakeEngineForTest(t))
bin := fakeEngineForTest(t)
manifest := testEngineManifest(bin)
platform := manifest.Platforms[hostKey()]
platform.Runtime.Env["FAKE_PID_FILE"] = pidFile
platform.Runtime.Env["FAKE_START_DELAY"] = "1m"
Expand Down Expand Up @@ -162,11 +163,7 @@ func TestE2EPrepareShutdownCancelsStartingEngine(t *testing.T) {
if pid == 0 {
t.Fatal("fake engine did not publish its PID")
}
t.Cleanup(func() {
if pidAlive(pid) {
_ = signalPID(pid, true)
}
})
t.Cleanup(func() { killOwnedTestPID(pid, bin) })

send(t, manager.stdin, 2, prepareShutdownMethod, nil)
waitResult(t, manager.frames, "2", 10*time.Second)
Expand Down
Loading
Loading