Support running host processes alongside VMs (#482)

* Regenerate .pb with protoc-gen-go v1.36.11

* Support running host processes alongside VMs

* Replace existing host processes when starting a set

internal/worker/hostprocess/process.go

@@ -0,0 +1,170 @@
+//nolint:err113,mnd // Preserve the original host-process errors and retry policy.
+package hostprocess
+
+import (
+	"context"
+	"errors"
+	"maps"
+	"net"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"strings"
+	"syscall"
+	"time"
+
+	"github.com/avast/retry-go/v4"
+	v1 "github.com/cirruslabs/orchard/pkg/resource/v1"
+)
+
+type Process struct {
+	cancel     context.CancelFunc
+	socketPath string
+	done       chan struct{}
+}
+
+func NewProcess(
+	spec v1.HostProcess,
+	workerName string,
+	vmName string,
+	controlSocket string,
+) (*Process, error) {
+	// Create a directory where the host process will create its Unix socket
+	//
+	// To keep the Unix socket path short enough and fit the platform limit,
+	// we're specifically requesting "/var/tmp" instead of "/var/folders/.../T/".
+	//
+	// And unlike "/tmp", "/var/tmp" is not subject to macOS's nightly cleanup.
+	runtimeDir, err := os.MkdirTemp("/var/tmp", "orchard-hp-")
+	if err != nil {
+		return nil, err
+	}
+
+	// Use a fixed socket name within the process-specific runtime directory
+	socketPath := filepath.Join(runtimeDir, "process.sock")
+
+	// Route the Tart control socket through the runtime directory as well
+	// to work around macOS's 104-byte Unix-domain socket limit
+	controlSocket, err = shortenControlSocketPath(runtimeDir, controlSocket)
+	if err != nil {
+		return nil, errors.Join(err, os.RemoveAll(runtimeDir))
+	}
+
+	// Give the process an independently cancellable lifetime
+	cmdCtx, cmdCtxCancel := context.WithCancel(context.Background())
+
+	//nolint:gosec // Executing the configured host process is the purpose of this API
+	command := exec.CommandContext(
+		cmdCtx,
+		spec.Program,
+		expandArgs(spec.Args, workerName, vmName, controlSocket, socketPath)...,
+	)
+	command.Dir = runtimeDir
+
+	// Ensure that we terminate the host process and any children it spawns
+	command.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
+	command.Cancel = func() error {
+		return syscall.Kill(-command.Process.Pid, syscall.SIGKILL)
+	}
+
+	// Craft the environment starting from the host process specification
+	env := make(map[string]string)
+
+	maps.Copy(env, spec.Env)
+
+	// Preserve only PATH from the worker and make Orchard-managed variables authoritative
+	env["PATH"] = os.Getenv("PATH")
+	env["ORCHARD_WORKER_NAME"] = workerName
+	env["ORCHARD_VM_NAME"] = vmName
+	env["ORCHARD_VM_CONTROL_SOCKET"] = controlSocket
+	env["ORCHARD_PROCESS_SOCKET"] = socketPath
+
+	// Convert the environment to the format expected by exec.Cmd
+	for name, value := range env {
+		command.Env = append(command.Env, name+"="+value)
+	}
+
+	if err := command.Start(); err != nil {
+		cmdCtxCancel()
+
+		return nil, errors.Join(err, os.RemoveAll(runtimeDir))
+	}
+
+	process := &Process{
+		cancel:     cmdCtxCancel,
+		socketPath: socketPath,
+		done:       make(chan struct{}),
+	}
+
+	go func() {
+		defer close(process.done)
+
+		// Reap the process; callers observe termination through done regardless of exit status
+		_ = command.Wait()
+		_ = command.Cancel()
+		_ = os.RemoveAll(runtimeDir)
+	}()
+
+	return process, nil
+}
+
+func shortenControlSocketPath(runtimeDir string, controlSocketPath string) (string, error) {
+	// Make the symlink target absolute because a relative TART_HOME would
+	// otherwise be resolved from runtimeDir, breaking the symlink
+	absoluteControlSocketPath, err := filepath.Abs(controlSocketPath)
+	if err != nil {
+		return "", err
+	}
+
+	// Place the VM's control socket alias in the runtime directory
+	aliasControlSocketPath := filepath.Join(runtimeDir, "vm.sock")
+
+	if err := os.Symlink(absoluteControlSocketPath, aliasControlSocketPath); err != nil {
+		return "", err
+	}
+
+	return aliasControlSocketPath, nil
+}
+
+func (process *Process) Dial(ctx context.Context) (net.Conn, error) {
+	var dialer net.Dialer
+
+	return retry.DoWithData(func() (net.Conn, error) {
+		select {
+		case <-process.done:
+			return nil, retry.Unrecoverable(errors.New("process exited before accepting connections"))
+		default:
+			// Continue
+		}
+
+		return dialer.DialContext(ctx, "unix", process.socketPath)
+	},
+		retry.Context(ctx),
+		retry.Attempts(200),
+		retry.Delay(50*time.Millisecond),
+		retry.DelayType(retry.FixedDelay),
+		retry.LastErrorOnly(true),
+	)
+}
+
+func (process *Process) Close() {
+	process.cancel()
+	<-process.done
+}
+
+func expandArgs(args []string, workerName string, vmName string, controlSocket string, socketPath string) []string {
+	replacer := strings.NewReplacer(
+		"${ORCHARD_WORKER_NAME}", workerName,
+		"${ORCHARD_VM_NAME}", vmName,
+		"${ORCHARD_VM_CONTROL_SOCKET}", controlSocket,
+		"${ORCHARD_PROCESS_SOCKET}", socketPath,
+	)
+
+	var expanded []string
+
+	for _, arg := range args {
+		expanded = append(expanded, replacer.Replace(arg))
+	}
+
+	return expanded
+}

internal/worker/hostprocess/process_test.go

@@ -0,0 +1,58 @@
+//nolint:noctx,testpackage,usetesting // Preserve the socket-path tests and their required short /var/tmp paths.
+package hostprocess
+
+import (
+	"net"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestShortenControlSocketPath(t *testing.T) {
+	// Create a control socket path that exceeds macOS's 104-byte limit
+	const controlSocketName = "control.sock"
+
+	vmDir := filepath.Join(t.TempDir(), strings.Repeat("v", 104))
+	require.NoError(t, os.MkdirAll(vmDir, 0o700))
+
+	controlSocketPath := filepath.Join(vmDir, controlSocketName)
+	require.Greater(t, len(controlSocketPath), 104)
+
+	// Create Orchard's short per-process runtime directory
+	runtimeDir, err := os.MkdirTemp("/var/tmp", "orchard-hp-test-")
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		require.NoError(t, os.RemoveAll(runtimeDir))
+	})
+
+	// Shorten the control socket path and verify the resulting alias
+	shortControlSocketPath, err := shortenControlSocketPath(runtimeDir, controlSocketPath)
+	require.NoError(t, err)
+	require.Equal(t, filepath.Join(runtimeDir, "vm.sock"), shortControlSocketPath)
+	require.Less(t, len(shortControlSocketPath), 104)
+
+	// Model Tart binding the socket relative to the long VM directory
+	t.Chdir(vmDir)
+
+	listener, err := net.Listen("unix", controlSocketName)
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		require.NoError(t, listener.Close())
+	})
+
+	// Model the host process running from Orchard's runtime directory
+	t.Chdir(runtimeDir)
+
+	// Verify connecting through the long absolute path fails
+	connection, err := net.Dial("unix", controlSocketPath)
+	require.Error(t, err)
+	require.Nil(t, connection)
+
+	// Connect to Tart's control socket through Orchard's short alias
+	connection, err = net.Dial("unix", shortControlSocketPath)
+	require.NoError(t, err)
+	require.NoError(t, connection.Close())
+}

internal/worker/hostprocess/set.go

@@ -0,0 +1,164 @@
+//nolint:contextcheck,err113 // Host processes have independent lifetimes; preserve the original errors.
+package hostprocess
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"sync"
+
+	"github.com/cirruslabs/orchard/internal/worker/ondiskname"
+	v1 "github.com/cirruslabs/orchard/pkg/resource/v1"
+)
+
+// Set owns the host processes associated with one VM.
+type Set struct {
+	workerName string
+	vmName     string
+	onDiskName ondiskname.OnDiskName
+	processes  map[string]*Process
+	started    bool
+	mtx        sync.Mutex
+}
+
+func NewSet(
+	workerName string,
+	vmName string,
+	onDiskName ondiskname.OnDiskName,
+) *Set {
+	return &Set{
+		workerName: workerName,
+		vmName:     vmName,
+		onDiskName: onDiskName,
+		processes:  make(map[string]*Process),
+	}
+}
+
+func (set *Set) Start(ctx context.Context, specs []v1.HostProcess) error {
+	set.mtx.Lock()
+	defer set.mtx.Unlock()
+
+	set.stopLocked()
+
+	return set.startLocked(ctx, specs)
+}
+
+func (set *Set) Replace(ctx context.Context, specs []v1.HostProcess) error {
+	set.mtx.Lock()
+	defer set.mtx.Unlock()
+
+	set.stopLocked()
+
+	return set.startLocked(ctx, specs)
+}
+
+func (set *Set) Dial(ctx context.Context, name string) (net.Conn, error) {
+	process := set.Lookup(name)
+
+	if process == nil {
+		return nil, fmt.Errorf("host process %q is not running", name)
+	}
+
+	connection, err := process.Dial(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("failed to connect to host process %q: %w", name, err)
+	}
+
+	return connection, nil
+}
+
+func (set *Set) Lookup(name string) *Process {
+	set.mtx.Lock()
+	defer set.mtx.Unlock()
+
+	return set.processes[name]
+}
+
+func (set *Set) Ready() bool {
+	set.mtx.Lock()
+	defer set.mtx.Unlock()
+
+	if !set.started {
+		return false
+	}
+
+	for _, process := range set.processes {
+		select {
+		case <-process.done:
+			return false
+		default:
+		}
+	}
+
+	return true
+}
+
+func (set *Set) Stop() {
+	set.mtx.Lock()
+	defer set.mtx.Unlock()
+
+	set.stopLocked()
+}
+
+func (set *Set) startLocked(ctx context.Context, specs []v1.HostProcess) error {
+	set.started = false
+
+	if len(specs) == 0 {
+		set.started = true
+
+		return nil
+	}
+
+	controlSocket, err := set.onDiskName.ControlSocketPath()
+	if err != nil {
+		return err
+	}
+
+	// Start and register every process
+	//
+	// If any process fails to start, stop the entire set
+	// rather than leaving it partially running.
+	for _, spec := range specs {
+		process, err := NewProcess(
+			spec,
+			set.workerName,
+			set.vmName,
+			controlSocket,
+		)
+		if err != nil {
+			set.stopLocked()
+
+			return fmt.Errorf("failed to start host process %q: %w", spec.Name, err)
+		}
+
+		set.processes[spec.Name] = process
+	}
+
+	// A host process is ready once it accepts connections on its socket
+	for _, spec := range specs {
+		process := set.processes[spec.Name]
+
+		connection, err := process.Dial(ctx)
+		if err != nil {
+			set.stopLocked()
+
+			return fmt.Errorf("failed to connect to host process %q: %w", spec.Name, err)
+		}
+
+		_ = connection.Close()
+	}
+
+	set.started = true
+
+	return nil
+}
+
+func (set *Set) stopLocked() {
+	set.started = false
+
+	for name, process := range set.processes {
+		delete(set.processes, name)
+
+		process.Close()
+	}
+}

internal/worker/hostprocess/set_test.go

@@ -0,0 +1,170 @@
+//nolint:goconst,noctx,testpackage // Preserve the original helper-process fixtures.
+package hostprocess
+
+import (
+	"net"
+	"os"
+	"testing"
+
+	"github.com/cirruslabs/orchard/internal/worker/ondiskname"
+	v1 "github.com/cirruslabs/orchard/pkg/resource/v1"
+	"github.com/stretchr/testify/require"
+)
+
+const (
+	testHelperArg          = "--orchard-host-process-test-helper"
+	testHelperEnv          = "ORCHARD_HOST_PROCESS_TEST_ENV"
+	testHelperInheritedEnv = "ORCHARD_HOST_PROCESS_TEST_INHERITED_ENV"
+)
+
+func TestMain(m *testing.M) {
+	if len(os.Args) == 2 && os.Args[1] == testHelperArg {
+		os.Exit(runTestHelper())
+	}
+
+	os.Exit(m.Run())
+}
+
+func TestSetStartAndStop(t *testing.T) {
+	t.Setenv(testHelperInheritedEnv, "must-not-be-inherited")
+
+	// Use this test binary as a host process
+	executable, err := os.Executable()
+	require.NoError(t, err)
+
+	// Create an empty host process set
+	set := NewSet("", "", ondiskname.OnDiskName{})
+	t.Cleanup(set.Stop)
+
+	// Ensure that a new set is not ready until started
+	require.False(t, set.Ready())
+
+	// Start the first host process
+	require.NoError(t, set.Start(t.Context(), []v1.HostProcess{{
+		Name:    "first",
+		Program: executable,
+		Args:    []string{testHelperArg},
+		Env: map[string]string{
+			testHelperEnv:            "set",
+			"PATH":                   "must-not-win",
+			"ORCHARD_PROCESS_SOCKET": "/must-not-win",
+		},
+	}}))
+	require.True(t, set.Ready())
+
+	// Ensure that the first host process is reachable
+	connection, err := set.Dial(t.Context(), "first")
+	require.NoError(t, err)
+	require.NoError(t, connection.Close())
+
+	// A failed replacement leaves the set not ready
+	require.Error(t, set.Replace(t.Context(), []v1.HostProcess{{
+		Name:    "invalid",
+		Program: "/does/not/exist",
+	}}))
+	require.False(t, set.Ready())
+	_, err = set.Dial(t.Context(), "first")
+	require.Error(t, err)
+
+	// Replace the first host process
+	require.NoError(t, set.Replace(t.Context(), []v1.HostProcess{{
+		Name:    "replacement",
+		Program: executable,
+		Args:    []string{testHelperArg},
+		Env: map[string]string{
+			testHelperEnv:            "set",
+			"PATH":                   "must-not-win",
+			"ORCHARD_PROCESS_SOCKET": "/must-not-win",
+		},
+	}}))
+	require.True(t, set.Ready())
+
+	// Ensure that only the replacement host process is reachable
+	_, err = set.Dial(t.Context(), "first")
+	require.Error(t, err)
+
+	connection, err = set.Dial(t.Context(), "replacement")
+	require.NoError(t, err)
+	require.NoError(t, connection.Close())
+
+	// Stop all host processes
+	set.Stop()
+	require.False(t, set.Ready())
+
+	// Ensure that the stopped host process is no longer reachable
+	_, err = set.Dial(t.Context(), "replacement")
+	require.Error(t, err)
+
+	// Ensure that an explicitly started empty set is ready
+	require.NoError(t, set.Start(t.Context(), nil))
+	require.True(t, set.Ready())
+}
+
+func TestSetStartReplacesExistingProcesses(t *testing.T) {
+	// Use this test binary as a host process
+	executable, err := os.Executable()
+	require.NoError(t, err)
+
+	// Create a host process set and start a new host process
+	set := NewSet("", "", ondiskname.OnDiskName{})
+	defer set.Stop()
+
+	hostProcesses := []v1.HostProcess{{
+		Name:    "process",
+		Program: executable,
+		Args:    []string{testHelperArg},
+		Env: map[string]string{
+			testHelperEnv: "set",
+		},
+	}}
+
+	require.NoError(t, set.Start(t.Context(), hostProcesses))
+
+	// Keep track of the original process to verify it is stopped
+	original := set.Lookup("process")
+	require.NotNil(t, original)
+	defer original.Close()
+
+	// Start a new host process
+	require.NoError(t, set.Start(t.Context(), hostProcesses))
+
+	// Ensure that the original process was stopped
+	select {
+	case <-original.done:
+		// It was stopped, nice
+	default:
+		require.FailNow(t, "the original process is still running")
+	}
+
+	// Ensure that a new ready process replaced the original
+	replacement := set.Lookup("process")
+	require.NotNil(t, replacement)
+	require.NotSame(t, original, replacement)
+	require.True(t, set.Ready())
+}
+
+func runTestHelper() int {
+	if os.Getenv(testHelperEnv) != "set" {
+		return 3
+	}
+	if os.Getenv(testHelperInheritedEnv) != "" {
+		return 4
+	}
+	if os.Getenv("PATH") == "must-not-win" {
+		return 5
+	}
+
+	listener, err := net.Listen("unix", os.Getenv("ORCHARD_PROCESS_SOCKET"))
+	if err != nil {
+		return 2
+	}
+	defer listener.Close()
+
+	for {
+		connection, err := listener.Accept()
+		if err != nil {
+			return 0
+		}
+		_ = connection.Close()
+	}
+}

internal/worker/ondiskname/ondiskname.go

@@ -3,9 +3,12 @@ package ondiskname
 import (
 	"errors"
 	"fmt"
-	v1 "github.com/cirruslabs/orchard/pkg/resource/v1"
+	"os"
+	"path/filepath"
 	"strconv"
 	"strings"
+
+	v1 "github.com/cirruslabs/orchard/pkg/resource/v1"
 )
 
 var (
@@ -81,3 +84,19 @@ func Parse(s string) (OnDiskName, error) {
 func (odn OnDiskName) String() string {
 	return fmt.Sprintf("%s-%s-%s-%d", prefix, odn.Name, odn.UID, odn.RestartCount)
 }
+
+func (odn OnDiskName) ControlSocketPath() (string, error) {
+	// Try user-overridden TART_HOME first
+	tartHome := os.Getenv("TART_HOME")
+	if tartHome == "" {
+		// Fall back to default TART_HOME
+		homeDir, err := os.UserHomeDir()
+		if err != nil {
+			return "", fmt.Errorf("failed to determine user home directory: %w", err)
+		}
+
+		tartHome = filepath.Join(homeDir, ".tart")
+	}
+
+	return filepath.Join(tartHome, "vms", odn.String(), "control.sock"), nil
+}

internal/worker/rpc.go

@@ -67,6 +67,7 @@ func (worker *Worker) watchRPC(ctx context.Context, operationCtx context.Context
 	}
 }
 
+//nolint:nestif,protogetter // Preserve the original host-process forwarding implementation.
 func (worker *Worker) handlePortForward(
 	ctx context.Context,
 	client rpc.ControllerClient,
@@ -87,49 +88,73 @@ func (worker *Worker) handlePortForward(
 		return
 	}
 
-	var host string
+	var targetConn net.Conn
 
-	if portForwardAction.VmUid == "" {
-		// Port-forwarding request to a worker
-		host = "localhost"
-	} else {
-		// Port-forwarding request to a VM, find that VM
-		vm, ok := lo.Find(worker.vmm.List(), func(item vmmanager.VM) bool {
-			return item.Resource().UID == portForwardAction.VmUid
-		})
-		if !ok {
-			worker.logger.Warnf("port forwarding failed: failed to get the VM: %v", err)
+	if target := portForwardAction.Target; target != nil {
+		// Sanity check
+		if portForwardAction.VmUid != "" || portForwardAction.Port != 0 {
+			worker.logger.Warn("port forwarding failed: target and legacy fields are mutually exclusive")
 
 			return
 		}
 
-		// Obtain VM's IP address
-		host, err = vm.IP(ctx)
+		// Retrieve the typed host process target, it's the only possible target right now
+		hostProcess := target.GetHostProcess()
+		if hostProcess == nil || hostProcess.VmUid == "" || hostProcess.Name == "" {
+			worker.logger.Warn("port forwarding failed: invalid or unsupported target")
+			return
+		}
+
+		// Dial host process
+		targetConn, err = worker.dialHostProcess(ctx, hostProcess.VmUid, hostProcess.Name)
 		if err != nil {
-			worker.logger.Warnf("port forwarding failed: failed to get VM's IP: %v", err)
+			worker.logger.Warnf("port forwarding failed: failed to connect to host process: %v", err)
+			return
+		}
+	} else {
+		var host string
+
+		if portForwardAction.VmUid == "" {
+			// Port-forwarding request to a worker
+			host = "localhost"
+		} else {
+			// Port-forwarding request to a VM, find that VM
+			vm, ok := lo.Find(worker.vmm.List(), func(item vmmanager.VM) bool {
+				return item.Resource().UID == portForwardAction.VmUid
+			})
+			if !ok {
+				worker.logger.Warnf("port forwarding failed: failed to get VM with UID %q",
+					portForwardAction.VmUid)
+
+				return
+			}
+
+			// Obtain VM's IP address
+			host, err = vm.IP(ctx)
+			if err != nil {
+				worker.logger.Warnf("port forwarding failed: failed to get VM's IP: %v", err)
+
+				return
+			}
+		}
+
+		// Connect to the VM's port
+		if worker.dialer != nil {
+			targetConn, err = worker.dialer.DialContext(ctx, "tcp",
+				fmt.Sprintf("%s:%d", host, portForwardAction.Port))
+		} else {
+			dialer := net.Dialer{}
+
+			targetConn, err = dialer.DialContext(ctx, "tcp",
+				fmt.Sprintf("%s:%d", host, portForwardAction.Port))
+		}
+		if err != nil {
+			worker.logger.Warnf("port forwarding failed: failed to connect to the VM: %v", err)
 
 			return
 		}
 	}
 
-	// Connect to the VM's port
-	var vmConn net.Conn
-
-	if worker.dialer != nil {
-		vmConn, err = worker.dialer.DialContext(ctx, "tcp",
-			fmt.Sprintf("%s:%d", host, portForwardAction.Port))
-	} else {
-		dialer := net.Dialer{}
-
-		vmConn, err = dialer.DialContext(ctx, "tcp",
-			fmt.Sprintf("%s:%d", host, portForwardAction.Port))
-	}
-	if err != nil {
-		worker.logger.Warnf("port forwarding failed: failed to connect to the VM: %v", err)
-
-		return
-	}
-
 	// Proxy bytes
 	grpcConn := &grpc_net_conn.Conn{
 		Stream:   stream,
@@ -143,7 +168,7 @@ func (worker *Worker) handlePortForward(
 		}),
 	}
 
-	_ = proxy.Connections(vmConn, grpcConn)
+	_ = proxy.Connections(targetConn, grpcConn)
 }
 
 func (worker *Worker) handleGetIP(

internal/worker/rpcv2.go

@@ -71,10 +71,26 @@ func (worker *Worker) handlePortForwardV2(ctx context.Context, portForward *v1.P
 	}
 }
 
+//nolint:err113,perfsprint // Preserve the original host-process forwarding errors.
 func (worker *Worker) handlePortForwardV2Inner(
 	ctx context.Context,
 	portForward *v1.PortForwardAction,
 ) (net.Conn, error) {
+	if target := portForward.Target; target != nil {
+		// Sanity check
+		if portForward.VMUID != "" || portForward.Port != 0 {
+			return nil, fmt.Errorf("target and legacy fields are mutually exclusive")
+		}
+
+		// Retrieve the typed host process target, it's the only possible target right now
+		if target.HostProcess == nil || target.HostProcess.VMUID == "" || target.HostProcess.Name == "" {
+			return nil, fmt.Errorf("invalid or unsupported target")
+		}
+
+		// Dial host process
+		return worker.dialHostProcess(ctx, target.HostProcess.VMUID, target.HostProcess.Name)
+	}
+
 	var host string
 	var err error
 
@@ -87,7 +103,7 @@ func (worker *Worker) handlePortForwardV2Inner(
 			return item.Resource().UID == portForward.VMUID
 		})
 		if !ok {
-			return nil, fmt.Errorf("failed to get the VM: %v", err)
+			return nil, fmt.Errorf("failed to get VM with UID %q", portForward.VMUID)
 		}
 
 		// Obtain VM's IP address
@@ -159,3 +175,28 @@ func (worker *Worker) handleGetIPV2Inner(
 
 	return ip, nil
 }
+
+//nolint:err113,ireturn // Preserve the original VM lookup helper required by host processes.
+func (worker *Worker) findVMByUID(uid string) (vmmanager.VM, error) {
+	vm, ok := lo.Find(worker.vmm.List(), func(item vmmanager.VM) bool {
+		return item.Resource().UID == uid
+	})
+	if !ok {
+		return nil, fmt.Errorf("VM with UID %q not found", uid)
+	}
+
+	if !vm.Started() {
+		return nil, fmt.Errorf("VM with UID %q is not running", uid)
+	}
+
+	return vm, nil
+}
+
+func (worker *Worker) dialHostProcess(ctx context.Context, vmUID string, name string) (net.Conn, error) {
+	vm, err := worker.findVMByUID(vmUID)
+	if err != nil {
+		return nil, err
+	}
+
+	return vm.HostProcessSet().Dial(ctx, name)
+}

internal/worker/vmmanager/base/base.go

@@ -15,9 +15,12 @@ import (
 	"github.com/avast/retry-go/v4"
 	"github.com/cirruslabs/orchard/internal/dialer"
 	"github.com/cirruslabs/orchard/internal/worker/endpoint"
+	"github.com/cirruslabs/orchard/internal/worker/hostprocess"
+	"github.com/cirruslabs/orchard/internal/worker/ondiskname"
 	"github.com/cirruslabs/orchard/pkg/client"
 	v1 "github.com/cirruslabs/orchard/pkg/resource/v1"
 	mapset "github.com/deckarep/golang-set/v2"
+	"github.com/samber/lo"
 	"go.uber.org/zap"
 	"golang.org/x/crypto/ssh"
 )
@@ -43,16 +46,18 @@ type VM struct {
 
 	statusMessage atomic.Pointer[string]
 	err           atomic.Pointer[error]
+	hostProcesses *hostprocess.Set
 	endpoints     *endpoint.Set
 
 	logger *zap.SugaredLogger
 }
 
-func NewVM(logger *zap.SugaredLogger) *VM {
+func NewVM(vmResource v1.VM, onDiskName ondiskname.OnDiskName, logger *zap.SugaredLogger) *VM {
 	return &VM{
-		conditions: mapset.NewSet(v1.ConditionTypeCloning),
-		endpoints:  endpoint.NewSet(logger),
-		logger:     logger,
+		conditions:    mapset.NewSet(v1.ConditionTypeCloning),
+		hostProcesses: hostprocess.NewSet(vmResource.Worker, vmResource.Name, onDiskName),
+		endpoints:     endpoint.NewSet(logger),
+		logger:        logger,
 	}
 }
 
@@ -60,10 +65,18 @@ func (vm *VM) EndpointSet() *endpoint.Set {
 	return vm.endpoints
 }
 
+func (vm *VM) HostProcessSet() *hostprocess.Set {
+	return vm.hostProcesses
+}
+
 func (vm *VM) SetStarted(val bool) {
 	vm.started.Store(val)
 }
 
+func (vm *VM) Started() bool {
+	return vm.started.Load()
+}
+
 func (vm *VM) Status() v1.VMStatus {
 	if vm.Err() != nil {
 		return v1.VMStatusFailed
@@ -113,6 +126,11 @@ func (vm *VM) Conditions() []v1.Condition {
 	// The worker must observe transitions before applying a new specification.
 	return []v1.Condition{
 		vm.conditionTypeToCondition(v1.ConditionTypeRunning),
+		{
+			Type: v1.ConditionTypeHostProcessesReady,
+			State: lo.Ternary(vm.HostProcessSet().Ready(),
+				v1.ConditionStateTrue, v1.ConditionStateFalse),
+		},
 		vm.conditionTypeToCondition(v1.ConditionTypeSuspending),
 		vm.conditionTypeToCondition(v1.ConditionTypeStopping),
 	}

internal/worker/vmmanager/synthetic/synthetic.go

@@ -2,6 +2,7 @@ package synthetic
 
 import (
 	"context"
+	"fmt"
 	"math/rand"
 	"strings"
 	"sync"
@@ -41,13 +42,15 @@ func NewVM(
 		"vm_restart_count", vmResource.RestartCount,
 	)
 
+	onDiskName := ondiskname.NewFromResource(vmResource)
 	vm := &VM{
-		onDiskName: ondiskname.NewFromResource(vmResource),
+		onDiskName: onDiskName,
 		resource:   vmResource,
 		ctx:        ctx,
 		cancel:     cancel,
 		logger:     logger,
-		VM:         base.NewVM(logger),
+
+		VM: base.NewVM(vmResource, onDiskName, logger),
 	}
 
 	vm.wg.Add(1)
@@ -121,6 +124,8 @@ func (vm *VM) Start(eventStreamer *client.EventStreamer) {
 }
 
 func (vm *VM) Suspend() <-chan error {
+	vm.HostProcessSet().Stop()
+
 	vm.EndpointSet().Stop()
 	errChan := make(chan error, 1)
 
@@ -136,6 +141,8 @@ func (vm *VM) IP(ctx context.Context) (string, error) {
 }
 
 func (vm *VM) Stop() <-chan error {
+	vm.HostProcessSet().Stop()
+
 	vm.EndpointSet().Stop()
 	errChan := make(chan error, 1)
 
@@ -160,6 +167,13 @@ func (vm *VM) Delete() error {
 func (vm *VM) run(ctx context.Context, eventStreamer *client.EventStreamer) {
 	defer vm.ConditionsSet().RemoveAll(v1.ConditionTypeRunning, v1.ConditionTypeSuspending, v1.ConditionTypeStopping)
 
+	if err := vm.HostProcessSet().Start(ctx, vm.resource.HostProcesses); err != nil {
+		vm.SetErr(fmt.Errorf("failed to start host processes: %w", err))
+
+		return
+	}
+	defer vm.HostProcessSet().Stop()
+
 	vm.EndpointSet().Start()
 	defer vm.EndpointSet().Stop()
 

internal/worker/vmmanager/tart/delete_test.go

@@ -98,7 +98,7 @@ func newVMForDelete(t *testing.T, script string) (*VM, string) {
 		logger:     logger,
 		ctx:        ctx,
 		cancel:     cancel,
-		VM:         base.NewVM(logger),
+		VM:         base.NewVM(v1.VM{}, ondiskname.OnDiskName{}, logger),
 	}
 	vm.ConditionsSet().Remove(v1.ConditionTypeCloning)
 

internal/worker/vmmanager/tart/tart.go

@@ -66,8 +66,9 @@ func NewVM(
 		"vm_restart_count", vmResource.RestartCount,
 	)
 
+	onDiskName := ondiskname.NewFromResource(vmResource)
 	vm := &VM{
-		onDiskName: ondiskname.NewFromResource(vmResource),
+		onDiskName: onDiskName,
 		resource:   vmResource,
 		logger:     logger,
 
@@ -79,7 +80,7 @@ func NewVM(
 		dialer:               dialer,
 		softnetPolicyUpdates: softnetPolicyUpdates,
 
-		VM: base.NewVM(logger),
+		VM: base.NewVM(vmResource, onDiskName, logger),
 	}
 
 	vm.wg.Add(1)
@@ -304,6 +305,12 @@ func (vm *VM) run(ctx context.Context, eventStreamer *client.EventStreamer) {
 	defer vm.ConditionsSet().RemoveAll(v1.ConditionTypeRunning, v1.ConditionTypeSuspending)
 
 	resource := vm.Resource()
+	if err := vm.HostProcessSet().Start(ctx, resource.HostProcesses); err != nil {
+		vm.SetErr(fmt.Errorf("failed to start host processes: %w", err))
+
+		return
+	}
+	defer vm.HostProcessSet().Stop()
 
 	vm.EndpointSet().Start()
 	defer vm.EndpointSet().Stop()
@@ -426,6 +433,8 @@ func (vm *VM) IP(ctx context.Context) (string, error) {
 }
 
 func (vm *VM) Suspend() <-chan error {
+	vm.HostProcessSet().Stop()
+
 	vm.EndpointSet().Stop()
 	errCh := make(chan error, 1)
 
@@ -459,6 +468,8 @@ func (vm *VM) Suspend() <-chan error {
 }
 
 func (vm *VM) Stop() <-chan error {
+	vm.HostProcessSet().Stop()
+
 	vm.EndpointSet().Stop()
 	vm.stopMtx.Lock()
 	defer vm.stopMtx.Unlock()

internal/worker/vmmanager/tart/tart_test.go

@@ -172,7 +172,7 @@ func newCloneTestVM(resource v1.VM) *VM {
 		onDiskName: ondiskname.NewFromResource(resource),
 		resource:   resource,
 		logger:     logger,
-		VM:         base.NewVM(logger),
+		VM:         base.NewVM(resource, ondiskname.NewFromResource(resource), logger),
 	}
 }
 

internal/worker/vmmanager/vetu/vetu.go

@@ -66,7 +66,7 @@ func NewVM(
 
 		dialer: dialer,
 
-		VM: base.NewVM(logger),
+		VM: base.NewVM(vmResource, ondiskname.NewFromResource(vmResource), logger),
 	}
 
 	vm.wg.Add(1)

internal/worker/vmmanager/vmmanager.go

@@ -4,6 +4,7 @@ import (
 	"context"
 
 	"github.com/cirruslabs/orchard/internal/worker/endpoint"
+	"github.com/cirruslabs/orchard/internal/worker/hostprocess"
 	"github.com/cirruslabs/orchard/internal/worker/ondiskname"
 	"github.com/cirruslabs/orchard/pkg/client"
 	v1 "github.com/cirruslabs/orchard/pkg/resource/v1"
@@ -17,6 +18,8 @@ type VM interface {
 	OnDiskName() ondiskname.OnDiskName
 	ImageFQN() *string
 	EndpointSet() *endpoint.Set
+	HostProcessSet() *hostprocess.Set
+	Started() bool
 	Status() v1.VMStatus
 	StatusMessage() string
 	Err() error

internal/worker/worker.go

@@ -663,6 +663,7 @@ func (worker *Worker) syncVMs(
 	return nil
 }
 
+//nolint:staticcheck // Preserve the original host-process specification comparison.
 func (worker *Worker) monitorRunningVM(
 	ctx context.Context,
 	vmResource *v1.VM,
@@ -674,6 +675,26 @@ func (worker *Worker) monitorRunningVM(
 	// Tracks whether any specification changes were applied without restarting the VM
 	appliedInPlace := false
 
+	// When set, retries during the next sync instead of restarting the VM
+	deferSpecReconciliation := false
+
+	// Try to apply the host process updates in-place or recover processes that exited unexpectedly
+	hostProcessesChanged := !currentVMResource.VMSpec.HostProcessesEqual(vmResource.VMSpec)
+	hostProcessesNeedRestart := currentVMResource.Generation == vmResource.Generation &&
+		v1.ConditionIsTrue(vm.Conditions(), v1.ConditionTypeRunning) &&
+		!vm.HostProcessSet().Ready()
+
+	if vmResource.PowerState == v1.PowerStateRunning && (hostProcessesChanged || hostProcessesNeedRestart) {
+		if err := vm.HostProcessSet().Replace(ctx, vmResource.HostProcesses); err != nil {
+			worker.logger.Warnf("failed to update host processes for VM %q: %v",
+				vmResource.Name, err)
+			deferSpecReconciliation = true
+		} else {
+			currentVMResource.HostProcesses = vmResource.HostProcesses
+			appliedInPlace = true
+		}
+	}
+
 	// Endpoint specification changes do not require restarting the VM;
 	// reconciliation happens separately below
 	endpointsChanged := !v1.SemanticallyEqual(
@@ -699,7 +720,10 @@ func (worker *Worker) monitorRunningVM(
 		vm.SetResource(currentVMResource)
 	}
 
-	worker.reconcileRunningVM(vmResource, vm) //nolint:contextcheck // Event streams outlive sync sessions.
+	// Reconcile specification changes that could not be applied in-place
+	if !deferSpecReconciliation {
+		worker.reconcileRunningVM(vmResource, vm) //nolint:contextcheck // Event streams outlive sync sessions.
+	}
 
 	var updateNeeded bool
 

internal/worker/worker_stop_test.go

@@ -87,7 +87,7 @@ func TestMonitorWaitsForStopBeforeApplyingGeneration(t *testing.T) {
 			require.NoError(t, err)
 			worker := &Worker{client: apiClient, runtime: runtime.NewSynthetic()}
 			vm := &delayedStopVM{
-				VM:          &synthetic.VM{VM: base.NewVM(zap.NewNop().Sugar())},
+				VM:          &synthetic.VM{VM: base.NewVM(v1.VM{}, ondiskname.OnDiskName{}, zap.NewNop().Sugar())},
 				resource:    v1.VM{Meta: v1.Meta{Name: "test-vm"}},
 				stopStarted: make(chan struct{}),
 				stopResult:  make(chan error),