Add endpoint support (#478)

* Add endpoint support

* Document endpoint network exposure

api/openapi.yaml

@@ -237,8 +237,10 @@ paths:
             application/json:
               schema:
                 $ref: '#/components/schemas/VM'
+        '400':
+          description: Invalid VM specification
         '409':
-          description: VM resource with with the same name already exists
+          description: VM resource with the same name already exists
     get:
       summary: "List VMs"
       tags:
@@ -311,6 +313,8 @@ paths:
             application/json:
               schema:
                 $ref: '#/components/schemas/VM'
+        '400':
+          description: Invalid VM specification
         '404':
           description: VM resource with the given name doesn't exist
     delete:
@@ -682,6 +686,19 @@ components:
           type: number
           description: Disk size for this VM
           example: 100
+        endpoints:
+          type: array
+          description: |
+            TCP services inside the VM to expose through TCP ports on the Orchard
+            Worker. The worker ports assigned to the endpoints are reported in
+            `observedEndpoints`.
+            Endpoint ports listen on all worker network interfaces. Use firewall
+            rules to restrict access.
+            VMs with endpoints are scheduled only on workers that support endpoint
+            forwarding. Adding endpoints to a VM already assigned to an unsupported
+            worker succeeds, but those endpoints are reported in the `error` state.
+          items:
+            $ref: '#/components/schemas/EndpointSpec'
         net-softnet:
           type: boolean
           description: Please use `netSoftnet` instead
@@ -854,6 +871,93 @@ components:
             Deprecated alias for `localName`.
           readOnly: true
           deprecated: true
+    ConnectionTarget:
+      title: Connection target
+      type: object
+      description: A connection destination relative to the containing resource or action.
+      required:
+        - vm
+      properties:
+        vm:
+          $ref: '#/components/schemas/ConnectionTargetVM'
+    ConnectionTargetVM:
+      title: VM connection target
+      type: object
+      required:
+        - port
+      properties:
+        port:
+          type: integer
+          minimum: 1
+          maximum: 65535
+          description: TCP port inside the contextual VM.
+    PortRange:
+      title: Port range
+      type: object
+      description: |
+        Inclusive bounds for selecting a network port. `min` must be less than or
+        equal to `max`; equal bounds identify one port.
+      required:
+        - min
+        - max
+      properties:
+        min:
+          type: integer
+          minimum: 1
+          maximum: 65535
+          description: Inclusive lower bound for port selection.
+        max:
+          type: integer
+          minimum: 1
+          maximum: 65535
+          description: Inclusive upper bound for port selection.
+    EndpointSpec:
+      title: Endpoint specification
+      type: object
+      description: A desired worker TCP endpoint backed by a connection target.
+      required:
+        - name
+        - target
+      properties:
+        name:
+          type: string
+          minLength: 1
+          description: Stable endpoint identifier, unique within `endpoints`.
+        target:
+          $ref: '#/components/schemas/ConnectionTarget'
+        workerPortRange:
+          description: |
+            Optional inclusive bounds for selecting the worker port. When omitted,
+            the operating system selects an available port. Equal bounds request
+            that exact port.
+          allOf:
+            - $ref: '#/components/schemas/PortRange'
+    EndpointStatus:
+      title: Endpoint status
+      type: object
+      description: Current observation for a desired endpoint.
+      required:
+        - name
+        - state
+      properties:
+        name:
+          type: string
+          minLength: 1
+          description: Stable endpoint identifier matching the desired endpoint.
+        workerPort:
+          type: integer
+          minimum: 1
+          maximum: 65535
+          description: TCP port assigned to this endpoint on the Orchard Worker.
+        state:
+          type: string
+          enum: [ listening, error ]
+          description: |
+            Exposure state. `listening` means that the worker TCP port is bound; it
+            does not imply that the service inside the VM is healthy or accepting connections.
+        message:
+          type: string
+          description: Human-readable detail, primarily populated when `state` is `error`.
     VMState:
       title: Virtual Machine State
       type: object
@@ -871,6 +975,13 @@ components:
         observedGeneration:
           type: number
           description: Corresponds to the `Generation` value on which the worker had acted upon
+        observedEndpoints:
+          type: array
+          readOnly: true
+          description: |
+            Current observations for the endpoints in `endpoints`.
+          items:
+            $ref: '#/components/schemas/EndpointStatus'
     Events:
       title: Events
       type: object

internal/controller/api_vms.go

@@ -43,6 +43,9 @@ func (controller *Controller) createVM(ctx *gin.Context) responder.Responder {
 	if vm.Image == "" {
 		return responder.JSON(http.StatusPreconditionFailed, NewErrorResponse("VM image is empty"))
 	}
+	if err := v1.ValidateEndpoints(vm.Endpoints); err != nil {
+		return responder.JSON(http.StatusBadRequest, NewErrorResponse("invalid endpoints: %v", err))
+	}
 
 	// Provide defaults
 	vm.Status = v1.VMStatusPending
@@ -56,6 +59,7 @@ func (controller *Controller) createVM(ctx *gin.Context) responder.Responder {
 	vm.TartName = vm.LocalName
 	vm.Generation = 0
 	vm.ObservedGeneration = 0
+	vm.ObservedEndpoints = nil
 	vm.Conditions = []v1.Condition{
 		{
 			Type:  v1.ConditionTypeScheduled,
@@ -158,6 +162,9 @@ func (controller *Controller) updateVMSpec(ctx *gin.Context) responder.Responder
 	if err := ctx.ShouldBindJSON(&userVM); err != nil {
 		return responder.JSON(http.StatusBadRequest, NewErrorResponse("invalid JSON was provided"))
 	}
+	if err := v1.ValidateEndpoints(userVM.Endpoints); err != nil {
+		return responder.JSON(http.StatusBadRequest, NewErrorResponse("invalid endpoints: %v", err))
+	}
 
 	if responder := controller.validateHostDirs(userVM.HostDirs); responder != nil {
 		return responder
@@ -221,7 +228,7 @@ func (controller *Controller) updateVMSpec(ctx *gin.Context) responder.Responder
 				"transition: only suspendable VMs can be suspended"))
 		}
 
-		if dbVM.SemanticallyEqual(userVM.VMSpec) {
+		if v1.SemanticallyEqual(dbVM.VMSpec, userVM.VMSpec) {
 			// Nothing was changed
 			return responder.JSON(http.StatusOK, dbVM)
 		}

internal/controller/api_workers.go

@@ -78,6 +78,7 @@ func (controller *Controller) createWorker(ctx *gin.Context) responder.Responder
 		dbWorker.Labels = worker.Labels
 		dbWorker.DefaultCPU = worker.DefaultCPU
 		dbWorker.DefaultMemory = worker.DefaultMemory
+		dbWorker.Capabilities = worker.Capabilities
 
 		if err := txn.SetWorker(*dbWorker); err != nil {
 			return responder.Error(err)

internal/controller/controller.go

@@ -332,6 +332,7 @@ func (controller *Controller) failVMsWithHostDirs() error {
 			vm.Status = v1.VMStatusFailed
 			vm.StatusMessage = "host directories are used, but host directory sharing is disabled"
 			vm.RestartPolicy = v1.RestartPolicyNever
+			vm.ObservedEndpoints = nil
 
 			if err := txn.SetVM(vm); err != nil {
 				return err

internal/controller/scheduler/scheduler.go

@@ -279,6 +279,12 @@ NextVM:
 				continue NextWorker
 			}
 
+			// Don't schedule VMs with endpoints on workers that don't support them
+			if len(unscheduledVM.Endpoints) != 0 &&
+				!worker.Capabilities.Has(v1.WorkerCapabilityVMEndpoints) {
+				continue NextWorker
+			}
+
 			err := scheduler.store.Update(func(txn storepkg.Transaction) error {
 				currentUnscheduledVM, err := txn.GetVM(unscheduledVM.Name)
 				if err != nil {
@@ -291,7 +297,8 @@ NextVM:
 					return err
 				}
 
-				if currentUnscheduledVM.UID != unscheduledVM.UID {
+				if currentUnscheduledVM.UID != unscheduledVM.UID ||
+					currentUnscheduledVM.Generation != unscheduledVM.Generation {
 					// The unscheduled VM had changed, so we'll re-evaluate a new
 					// version of it in the next scheduling loop iteration
 					return ErrVMSchedulingSkipped
@@ -333,6 +340,12 @@ NextVM:
 					return ErrWorkerSchedulingSkipped
 				}
 
+				// Don't schedule VMs with endpoints on workers that don't support them
+				if len(unscheduledVM.Endpoints) != 0 &&
+					!currentWorker.Capabilities.Has(v1.WorkerCapabilityVMEndpoints) {
+					return ErrWorkerSchedulingSkipped
+				}
+
 				if currentWorker.MachineID != worker.MachineID ||
 					!currentWorker.Resources.Equal(worker.Resources) {
 					// Worker has changed
@@ -504,6 +517,11 @@ func (scheduler *Scheduler) healthCheckingLoopIteration() (int, error) {
 func (scheduler *Scheduler) healthCheckVM(txn storepkg.Transaction, vm v1.VM) error {
 	logger := scheduler.logger.With("vm_name", vm.Name, "vm_uid", vm.UID, "vm_restart_count", vm.RestartCount)
 
+	// Preserve the current endpoint observations for comparison below and reset them,
+	// so every VM transition persisted below discards the stale endpoint observations
+	savedObservedEndpoints := vm.ObservedEndpoints
+	vm.ObservedEndpoints = nil
+
 	// Schedule a VM restart if the restart policy mandates it
 	needsRestart := vm.RestartPolicy == v1.RestartPolicyOnFailure &&
 		vm.Status == v1.VMStatusFailed &&
@@ -586,5 +604,24 @@ func (scheduler *Scheduler) healthCheckVM(txn storepkg.Transaction, vm v1.VM) er
 		return txn.SetVM(vm)
 	}
 
+	// Report unsupported endpoints to clients without failing the VM
+	if !worker.Capabilities.Has(v1.WorkerCapabilityVMEndpoints) {
+		vm.ObservedEndpoints = make([]v1.EndpointStatus, 0, len(vm.Endpoints))
+
+		for _, endpoint := range vm.Endpoints {
+			vm.ObservedEndpoints = append(vm.ObservedEndpoints, v1.EndpointStatus{
+				Name:    endpoint.Name,
+				State:   v1.EndpointStateError,
+				Message: "worker doesn't support VM endpoints",
+			})
+		}
+
+		if slices.Equal(savedObservedEndpoints, vm.ObservedEndpoints) {
+			return nil
+		}
+
+		return txn.SetVM(vm)
+	}
+
 	return nil
 }

internal/proxy/proxy.go

@@ -1,11 +1,28 @@
 package proxy
 
 import (
+	"errors"
 	"io"
+	"net"
 	"strings"
 )
 
+// halfCloseConnection can stop writing without interrupting concurrent reads.
+type halfCloseConnection interface {
+	net.Conn
+	CloseWrite() error
+}
+
+// Connections copies bytes in both directions. When both connections support
+// write-side shutdown, EOF is propagated independently in each direction.
+// Other connection pairs retain the historical close-on-first-completion behavior.
 func Connections(left io.ReadWriteCloser, right io.ReadWriteCloser) (finalErr error) {
+	if left, ok := left.(halfCloseConnection); ok {
+		if right, ok := right.(halfCloseConnection); ok {
+			return connectionsWithHalfClose(left, right)
+		}
+	}
+
 	leftErrCh := make(chan error, 1)
 	rightErrCh := make(chan error, 1)
 
@@ -44,3 +61,51 @@ func Connections(left io.ReadWriteCloser, right io.ReadWriteCloser) (finalErr er
 
 	return finalErr
 }
+
+// connectionsWithHalfClose keeps the reverse direction open after a clean EOF.
+func connectionsWithHalfClose(left, right halfCloseConnection) error {
+	//nolint:mnd // one result from each copy direction
+	results := make(chan error, 2)
+
+	copyHalf := func(destination, source halfCloseConnection) {
+		_, err := io.Copy(destination, source)
+		if err == nil {
+			err = destination.CloseWrite()
+		}
+
+		results <- err
+	}
+
+	closeBoth := func() {
+		_ = left.Close()
+		_ = right.Close()
+	}
+
+	meaningfulError := func(err error) error {
+		if err == nil ||
+			errors.Is(err, net.ErrClosed) ||
+			strings.Contains(err.Error(), "use of closed network connection") {
+			return nil
+		}
+
+		return err
+	}
+
+	go copyHalf(right, left)
+	go copyHalf(left, right)
+
+	firstErr := <-results
+	if firstErr != nil {
+		// An I/O or CloseWrite failure must unblock both copy operations.
+		closeBoth()
+	}
+
+	secondErr := <-results
+	closeBoth()
+
+	if err := meaningfulError(firstErr); err != nil {
+		return err
+	}
+
+	return meaningfulError(secondErr)
+}

internal/tests/endpoint_test.go

@@ -0,0 +1,89 @@
+//go:build darwin
+
+package tests_test
+
+import (
+	"net"
+	"strconv"
+	"testing"
+	"time"
+
+	"github.com/cirruslabs/orchard/internal/imageconstant"
+	"github.com/cirruslabs/orchard/internal/tests/devcontroller"
+	v1 "github.com/cirruslabs/orchard/pkg/resource/v1"
+	"github.com/stretchr/testify/require"
+	"golang.org/x/crypto/ssh"
+)
+
+func TestEndpoint(t *testing.T) {
+	ctx := t.Context()
+
+	devClient, _, _ := devcontroller.StartIntegrationTestEnvironment(t)
+
+	// Create a VM that exposes its SSH service through the worker endpoint
+	const (
+		vmName       = "endpoint-test-vm"
+		endpointName = "ssh"
+	)
+
+	require.NoError(t, devClient.VMs().Create(ctx, &v1.VM{
+		Name:     vmName,
+		Image:    imageconstant.DefaultMacosImage,
+		CPU:      4,
+		Memory:   8 * 1024,
+		Headless: true,
+		Endpoints: []v1.EndpointSpec{
+			{
+				Name: endpointName,
+				Target: v1.ConnectionTarget{
+					VM: &v1.ConnectionTargetVM{Port: 22},
+				},
+			},
+		},
+	}))
+
+	// Wait for the Worker to expose a listening endpoint on the running VM
+	var (
+		vm  *v1.VM
+		err error
+	)
+
+	require.Eventually(t, func() bool {
+		vm, err = devClient.VMs().Get(ctx, vmName)
+
+		return err == nil &&
+			vm.Status == v1.VMStatusRunning &&
+			len(vm.ObservedEndpoints) == 1 &&
+			vm.ObservedEndpoints[0].State == v1.EndpointStateListening
+	}, 2*time.Minute, time.Second, "failed to wait for the endpoint")
+
+	// Verify the endpoint uses the expected name and a dynamically assigned Worker port
+	endpointStatus := vm.ObservedEndpoints[0]
+	require.Equal(t, endpointName, endpointStatus.Name)
+	require.NotZero(t, endpointStatus.WorkerPort)
+
+	// Wait for SSH to accept connections through the worker port
+	var sshClient *ssh.Client
+	address := net.JoinHostPort("127.0.0.1", strconv.Itoa(int(endpointStatus.WorkerPort)))
+	sshConfig := &ssh.ClientConfig{
+		User:            "admin",
+		Auth:            []ssh.AuthMethod{ssh.Password("admin")},
+		HostKeyCallback: ssh.InsecureIgnoreHostKey(),
+		Timeout:         10 * time.Second,
+	}
+
+	require.Eventually(t, func() bool {
+		sshClient, err = ssh.Dial("tcp", address, sshConfig)
+		return err == nil
+	}, 2*time.Minute, time.Second, "failed to connect to the endpoint over SSH")
+	defer sshClient.Close()
+
+	// Run a command to verify the forwarded connection reaches the VM
+	sshSession, err := sshClient.NewSession()
+	require.NoError(t, err)
+	defer sshSession.Close()
+
+	unameOutput, err := sshSession.Output("uname -a")
+	require.NoError(t, err)
+	require.Contains(t, string(unameOutput), "Darwin")
+}

internal/worker/endpoint/endpoint.go

@@ -0,0 +1,197 @@
+package endpoint
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"sync/atomic"
+
+	"github.com/cirruslabs/orchard/internal/proxy"
+	v1 "github.com/cirruslabs/orchard/pkg/resource/v1"
+	"go.uber.org/zap"
+)
+
+// Bound file descriptor usage by a single endpoint.
+const maxEndpointListenerConnections = 128
+
+// Dial connects to a target after an endpoint accepts a connection.
+type Dial func(context.Context) (net.Conn, error)
+
+//nolint:containedctx // the listener and accepted connections share an owned cancellation lifetime
+type endpoint struct {
+	port            uint16
+	spec            v1.EndpointSpec
+	listener        net.Listener
+	dial            Dial
+	logger          *zap.SugaredLogger
+	failure         atomic.Pointer[error]
+	connectionSlots chan struct{}
+
+	ctx    context.Context
+	cancel context.CancelFunc
+}
+
+func newEndpoint(
+	spec v1.EndpointSpec,
+	bindTarget BindTarget,
+	logger *zap.SugaredLogger,
+) (*endpoint, error) {
+	// Prepare the target dialer before opening the worker listener
+	dial, err := bindTarget(spec.Target)
+	if err != nil {
+		return nil, err
+	}
+
+	// Claim a worker port from the requested range
+	listener, port, err := listen(spec.WorkerPortRange)
+	if err != nil {
+		return nil, err
+	}
+
+	// Give the listener and all accepted connections one shared lifetime
+	ctx, cancel := context.WithCancel(context.Background())
+	result := &endpoint{
+		port:            port,
+		spec:            spec,
+		listener:        listener,
+		dial:            dial,
+		logger:          logger,
+		ctx:             ctx,
+		cancel:          cancel,
+		connectionSlots: make(chan struct{}, maxEndpointListenerConnections),
+	}
+
+	// Begin accepting connections only after construction is complete
+	go result.accept()
+
+	return result, nil
+}
+
+//nolint:forcetypeassert,gosec,noctx // owned TCP listeners intentionally bind all interfaces and return TCPAddr
+func listen(portRange *v1.PortRange) (net.Listener, uint16, error) {
+	// Let the operating system select a port when no range was requested
+	if portRange == nil {
+		listener, err := net.Listen("tcp", ":0")
+		if err != nil {
+			return nil, 0, fmt.Errorf("failed to bind a TCP listener: %w", err)
+		}
+
+		return listener, uint16(listener.Addr().(*net.TCPAddr).Port), nil
+	}
+
+	var lastErr error
+
+	// Try every requested port in ascending order until one is available
+	for candidate := int(portRange.Min); candidate <= int(portRange.Max); candidate++ {
+		listener, err := net.Listen("tcp", fmt.Sprintf(":%d", candidate))
+		if err != nil {
+			lastErr = err
+
+			continue
+		}
+
+		return listener, uint16(candidate), nil
+	}
+
+	return nil, 0, fmt.Errorf(
+		"failed to bind a TCP listener in worker port range %d-%d: %w",
+		portRange.Min,
+		portRange.Max,
+		lastErr,
+	)
+}
+
+func (ep *endpoint) running() bool {
+	return ep.ctx.Err() == nil
+}
+
+func (ep *endpoint) status() v1.EndpointStatus {
+	if failure := ep.failure.Load(); failure != nil {
+		return v1.EndpointStatus{
+			Name:    ep.spec.Name,
+			State:   v1.EndpointStateError,
+			Message: (*failure).Error(),
+		}
+	}
+
+	return v1.EndpointStatus{
+		Name:       ep.spec.Name,
+		WorkerPort: ep.port,
+		State:      v1.EndpointStateListening,
+	}
+}
+
+func (ep *endpoint) fail(err error) {
+	// Preserve the first fatal listener error and make failure idempotent
+	if ep.failure.CompareAndSwap(nil, &err) {
+		ep.logger.Warnf("endpoint %q failed: %v", ep.spec.Name, err)
+		ep.close()
+	}
+}
+
+func (ep *endpoint) accept() {
+	// Accept connections until the endpoint stops or the listener fails
+	for {
+		select {
+		case ep.connectionSlots <- struct{}{}:
+			// Successfully obtained a connection slot, proceed
+		case <-ep.ctx.Done():
+			return
+		}
+
+		connection, err := ep.listener.Accept()
+		if err != nil {
+			// Return connection slot back
+			<-ep.connectionSlots
+
+			// Listener closure is expected during normal endpoint shutdown
+			if ep.running() {
+				ep.fail(fmt.Errorf("failed to accept connections: %w", err))
+			}
+
+			return
+		}
+
+		// Forward each accepted connection independently
+		go ep.forward(connection)
+	}
+}
+
+func (ep *endpoint) forward(connection net.Conn) {
+	// Return connection slot back once done
+	defer func() { <-ep.connectionSlots }()
+
+	// Close the client connection when forwarding ends or the endpoint stops
+	defer connection.Close()
+	stopClosingConnection := context.AfterFunc(ep.ctx, func() {
+		_ = connection.Close()
+	})
+	defer stopClosingConnection()
+
+	// Resolve and connect to the endpoint target lazily for this connection
+	targetConnection, err := ep.dial(ep.ctx)
+	if err != nil {
+		if ep.running() {
+			ep.logger.Debugf("failed to connect endpoint %q to its target: %v", ep.spec.Name, err)
+		}
+
+		return
+	}
+
+	// Close the target connection when forwarding ends or the endpoint stops
+	defer targetConnection.Close()
+	stopClosingTarget := context.AfterFunc(ep.ctx, func() {
+		_ = targetConnection.Close()
+	})
+	defer stopClosingTarget()
+
+	// Relay traffic in both directions until either side finishes
+	if err := proxy.Connections(connection, targetConnection); err != nil && ep.running() {
+		ep.logger.Debugf("endpoint %q TCP relay failed: %v", ep.spec.Name, err)
+	}
+}
+
+func (ep *endpoint) close() {
+	ep.cancel()
+	_ = ep.listener.Close()
+}

internal/worker/endpoint/endpoint_test.go

@@ -0,0 +1,185 @@
+//nolint:testpackage // exercises private listener lifecycle and relay behavior
+package endpoint
+
+import (
+	"context"
+	"io"
+	"net"
+	"testing"
+	"time"
+
+	v1 "github.com/cirruslabs/orchard/pkg/resource/v1"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/zap"
+)
+
+//nolint:forcetypeassert,noctx // the test owns its TCP listener and closes it through cleanup
+func TestEndpointPropagatesTCPHalfClose(t *testing.T) {
+	// Listen for the endpoint's target connection
+	backendListener, err := net.Listen("tcp", "127.0.0.1:0")
+	require.NoError(t, err)
+	t.Cleanup(func() { require.NoError(t, backendListener.Close()) })
+
+	// Consume one request through EOF before sending the backend response
+	backendResult := make(chan error, 1)
+
+	const (
+		request  = "request that ends at EOF"
+		response = "response after request EOF"
+	)
+
+	go func() {
+		backendConnection, err := backendListener.Accept()
+		if err != nil {
+			backendResult <- err
+			return
+		}
+		defer backendConnection.Close()
+
+		_, err = io.Copy(io.Discard, backendConnection)
+		if err == nil {
+			_, err = io.WriteString(backendConnection, response)
+		}
+
+		backendResult <- err
+	}()
+
+	// Create an endpoint that forwards accepted connections to the backend
+	bindTarget := func(v1.ConnectionTarget) (Dial, error) {
+		return func(ctx context.Context) (net.Conn, error) {
+			return (&net.Dialer{}).DialContext(ctx, "tcp", backendListener.Addr().String())
+		}, nil
+	}
+
+	ep, err := newEndpoint(v1.EndpointSpec{Name: "half-close"}, bindTarget, zap.NewNop().Sugar())
+	require.NoError(t, err)
+	t.Cleanup(ep.close)
+
+	// Connect a client using the address family selected by the endpoint listener
+	listenerAddress := ep.listener.Addr().(*net.TCPAddr)
+	clientAddress := &net.TCPAddr{IP: net.IPv6loopback, Port: listenerAddress.Port}
+	if listenerAddress.IP.To4() != nil {
+		clientAddress.IP = net.IPv4(127, 0, 0, 1)
+	}
+
+	client, err := net.DialTCP("tcp", nil, clientAddress)
+	require.NoError(t, err)
+	defer client.Close()
+	require.NoError(t, client.SetDeadline(time.Now().Add(5*time.Second)))
+
+	// End the request with a half-close while keeping the client read side open
+	_, err = io.WriteString(client, request)
+	require.NoError(t, err)
+	require.NoError(t, client.CloseWrite())
+
+	// Verify the backend response crosses the still-open reverse direction
+	received, err := io.ReadAll(client)
+	require.NoError(t, err)
+	require.Equal(t, response, string(received))
+	require.NoError(t, <-backendResult)
+}
+
+func TestSetRecreatesFailedEndpoint(t *testing.T) {
+	endpointSet := NewSet(zap.NewNop().Sugar())
+	endpointSet.Start()
+	t.Cleanup(endpointSet.Stop)
+
+	const endpointName = "ssh"
+
+	// Create a listening endpoint
+	endpointSpecs := []v1.EndpointSpec{{Name: endpointName}}
+
+	statuses := endpointSet.Reconcile(endpointSpecs, testBindTarget)
+	require.Len(t, statuses, 1)
+	require.Equal(t, v1.EndpointStateListening, statuses[0].State)
+
+	// Simulate an unexpected listener failure
+	originalEndpoint := endpointSet.endpoints[endpointName]
+	require.NotNil(t, originalEndpoint)
+	require.NoError(t, originalEndpoint.listener.Close())
+
+	// Wait for the endpoint to report the listener failure
+	require.Eventually(t, func() bool {
+		return originalEndpoint.status().State == v1.EndpointStateError
+	}, time.Second, 10*time.Millisecond)
+
+	failedStatus := originalEndpoint.status()
+	require.Zero(t, failedStatus.WorkerPort)
+	require.Contains(t, failedStatus.Message, "failed to accept connections")
+
+	// Reconcile again and verify the failed endpoint is replaced
+	statuses = endpointSet.Reconcile(endpointSpecs, testBindTarget)
+	require.Len(t, statuses, 1)
+	require.Equal(t, v1.EndpointStateListening, statuses[0].State)
+	require.NotSame(t, originalEndpoint, endpointSet.endpoints[endpointName])
+}
+
+func TestSetRecreatesEndpointsWhenDesiredSetChanges(t *testing.T) {
+	endpointSet := NewSet(zap.NewNop().Sugar())
+	endpointSet.Start()
+	t.Cleanup(endpointSet.Stop)
+
+	// Start with one endpoint whose worker port can move
+	flexibleSpec := v1.EndpointSpec{Name: "flexible"}
+	statuses := endpointSet.Reconcile([]v1.EndpointSpec{flexibleSpec}, testBindTarget)
+	require.Len(t, statuses, 1)
+	require.Equal(t, v1.EndpointStateListening, statuses[0].State)
+
+	claimedPort := statuses[0].WorkerPort
+	originalFlexible := endpointSet.endpoints[flexibleSpec.Name]
+
+	// Add a fixed-port endpoint first, forcing the whole set to be reallocated
+	statuses = endpointSet.Reconcile(
+		[]v1.EndpointSpec{
+			{
+				Name:            "fixed",
+				WorkerPortRange: &v1.PortRange{Min: claimedPort, Max: claimedPort},
+			},
+			flexibleSpec,
+		},
+		testBindTarget,
+	)
+
+	// Verify the fixed endpoint takes the old port and the flexible endpoint moves
+	require.Len(t, statuses, 2)
+	require.Equal(t, v1.EndpointStateListening, statuses[0].State)
+	require.Equal(t, claimedPort, statuses[0].WorkerPort)
+	require.Equal(t, v1.EndpointStateListening, statuses[1].State)
+	require.NotEqual(t, claimedPort, statuses[1].WorkerPort)
+	require.NotSame(t, originalFlexible, endpointSet.endpoints[flexibleSpec.Name])
+}
+
+func TestSetDoesNotRepeatFullResetAfterPartialFailure(t *testing.T) {
+	// Occupy a worker port so one desired endpoint cannot start
+	blocker, blockedPort, err := listen(nil)
+	require.NoError(t, err)
+	t.Cleanup(func() { require.NoError(t, blocker.Close()) })
+
+	endpointSet := NewSet(zap.NewNop().Sugar())
+	endpointSet.Start()
+	t.Cleanup(endpointSet.Stop)
+
+	desired := []v1.EndpointSpec{
+		{
+			Name:            "blocked",
+			WorkerPortRange: &v1.PortRange{Min: blockedPort, Max: blockedPort},
+		},
+		{Name: "healthy"},
+	}
+
+	// Apply the desired set once and remember its healthy listener
+	endpointSet.Reconcile(desired, testBindTarget)
+	require.NotContains(t, endpointSet.endpoints, "blocked")
+	healthyEndpoint := endpointSet.endpoints["healthy"]
+	require.NotNil(t, healthyEndpoint)
+
+	// Reconcile unchanged desired state and verify the healthy listener is preserved
+	endpointSet.Reconcile(desired, testBindTarget)
+	require.Same(t, healthyEndpoint, endpointSet.endpoints["healthy"])
+}
+
+func testBindTarget(v1.ConnectionTarget) (Dial, error) {
+	return func(context.Context) (net.Conn, error) {
+		return nil, net.ErrClosed
+	}, nil
+}

internal/worker/endpoint/set.go

@@ -0,0 +1,108 @@
+package endpoint
+
+import (
+	"slices"
+	"sync"
+
+	v1 "github.com/cirruslabs/orchard/pkg/resource/v1"
+	"go.uber.org/zap"
+)
+
+// Set owns the endpoint listeners associated with one VM.
+type Set struct {
+	logger      *zap.SugaredLogger
+	endpoints   map[string]*endpoint
+	lastDesired []v1.EndpointSpec
+	started     bool
+	mtx         sync.Mutex
+}
+
+func NewSet(logger *zap.SugaredLogger) *Set {
+	return &Set{
+		logger:    logger,
+		endpoints: make(map[string]*endpoint),
+	}
+}
+
+func (set *Set) Start() {
+	set.mtx.Lock()
+	defer set.mtx.Unlock()
+
+	set.stopLocked()
+	set.started = true
+}
+
+func (set *Set) Reconcile(
+	desired []v1.EndpointSpec,
+	bindTarget BindTarget,
+) []v1.EndpointStatus {
+	set.mtx.Lock()
+	defer set.mtx.Unlock()
+
+	// Ignore reconciliation until the owning VM starts the endpoint set
+	if !set.started {
+		return nil
+	}
+
+	// Recreate the entire endpoint set on any change to avoid conflicts
+	// between current and desired worker-port assignments
+	if !v1.SemanticallyEqual(set.lastDesired, desired) {
+		set.stopLocked()
+		set.lastDesired = slices.Clone(desired)
+	}
+
+	if len(desired) == 0 {
+		return nil
+	}
+
+	// Reuse healthy unchanged endpoints and recreate every other endpoint
+	statuses := make([]v1.EndpointStatus, 0, len(desired))
+
+	for _, spec := range desired {
+		name := spec.Name
+
+		if current := set.endpoints[name]; current != nil {
+			status := current.status()
+
+			if status.State == v1.EndpointStateListening {
+				statuses = append(statuses, status)
+
+				continue
+			}
+
+			delete(set.endpoints, name)
+			current.close()
+		}
+
+		current, err := newEndpoint(spec, bindTarget, set.logger)
+		if err != nil {
+			statuses = append(statuses, v1.EndpointStatus{
+				Name:    name,
+				State:   v1.EndpointStateError,
+				Message: err.Error(),
+			})
+
+			continue
+		}
+
+		set.endpoints[name] = current
+		statuses = append(statuses, current.status())
+	}
+
+	return statuses
+}
+
+func (set *Set) Stop() {
+	set.mtx.Lock()
+	defer set.mtx.Unlock()
+
+	set.started = false
+	set.stopLocked()
+}
+
+func (set *Set) stopLocked() {
+	for _, current := range set.endpoints {
+		current.close()
+	}
+	clear(set.endpoints)
+}

internal/worker/endpoint/target.go

@@ -0,0 +1,69 @@
+package endpoint
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"strconv"
+
+	"github.com/cirruslabs/orchard/internal/dialer"
+	v1 "github.com/cirruslabs/orchard/pkg/resource/v1"
+	"golang.org/x/sync/singleflight"
+)
+
+// BindTarget validates a declarative target and returns a lazy Dial.
+// It runs synchronously during reconciliation and must not perform I/O.
+type BindTarget func(v1.ConnectionTarget) (Dial, error)
+
+//nolint:err113,forcetypeassert,perfsprint // preserve validation errors; the coalesced IP resolver returns a string
+func NewVMTargetBinder(
+	resolveIP func(context.Context) (string, error),
+	networkDialer dialer.Dialer,
+) BindTarget {
+	// Use the standard network dialer unless the caller supplied one
+	if networkDialer == nil {
+		networkDialer = &net.Dialer{}
+	}
+
+	return func(target v1.ConnectionTarget) (Dial, error) {
+		// Validate the target synchronously during endpoint reconciliation
+		if err := target.Validate(); err != nil {
+			return nil, fmt.Errorf("invalid connection target: %w", err)
+		}
+		if target.VM == nil {
+			return nil, fmt.Errorf("unsupported connection target")
+		}
+
+		// Capture the target port by value for the lifetime of this endpoint
+		targetPort := target.VM.Port
+
+		// Coalesce concurrent IP lookups for this endpoint without caching results
+		var resolveGroup singleflight.Group
+
+		return func(ctx context.Context) (net.Conn, error) {
+			// Resolve the VM address lazily because it can change across VM runs
+			resolved, err, _ := resolveGroup.Do("vm-ip", func() (any, error) {
+				return resolveIP(ctx)
+			})
+			if err != nil {
+				return nil, fmt.Errorf("failed to get VM's IP: %w", err)
+			}
+
+			// Reject an empty host before address construction can reinterpret it
+			host := resolved.(string)
+			if host == "" {
+				return nil, fmt.Errorf("failed to get VM's IP: empty address")
+			}
+
+			address := net.JoinHostPort(host, strconv.Itoa(int(targetPort)))
+
+			// Connect to the resolved VM address using the caller's cancellation context
+			connection, err := networkDialer.DialContext(ctx, "tcp", address)
+			if err != nil {
+				return nil, fmt.Errorf("failed to connect to the VM: %w", err)
+			}
+
+			return connection, nil
+		}, nil
+	}
+}

internal/worker/vmmanager/base/base.go

@@ -14,6 +14,7 @@ import (
 
 	"github.com/avast/retry-go/v4"
 	"github.com/cirruslabs/orchard/internal/dialer"
+	"github.com/cirruslabs/orchard/internal/worker/endpoint"
 	"github.com/cirruslabs/orchard/pkg/client"
 	v1 "github.com/cirruslabs/orchard/pkg/resource/v1"
 	mapset "github.com/deckarep/golang-set/v2"
@@ -42,6 +43,7 @@ type VM struct {
 
 	statusMessage atomic.Pointer[string]
 	err           atomic.Pointer[error]
+	endpoints     *endpoint.Set
 
 	logger *zap.SugaredLogger
 }
@@ -49,10 +51,15 @@ type VM struct {
 func NewVM(logger *zap.SugaredLogger) *VM {
 	return &VM{
 		conditions: mapset.NewSet(v1.ConditionTypeCloning),
+		endpoints:  endpoint.NewSet(logger),
 		logger:     logger,
 	}
 }
 
+func (vm *VM) EndpointSet() *endpoint.Set {
+	return vm.endpoints
+}
+
 func (vm *VM) SetStarted(val bool) {
 	vm.started.Store(val)
 }

internal/worker/vmmanager/synthetic/synthetic.go

@@ -121,6 +121,7 @@ func (vm *VM) Start(eventStreamer *client.EventStreamer) {
 }
 
 func (vm *VM) Suspend() <-chan error {
+	vm.EndpointSet().Stop()
 	errChan := make(chan error, 1)
 
 	errChan <- nil
@@ -135,6 +136,7 @@ func (vm *VM) IP(ctx context.Context) (string, error) {
 }
 
 func (vm *VM) Stop() <-chan error {
+	vm.EndpointSet().Stop()
 	errChan := make(chan error, 1)
 
 	errChan <- nil
@@ -158,6 +160,9 @@ 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)
 
+	vm.EndpointSet().Start()
+	defer vm.EndpointSet().Stop()
+
 	// Launch the startup script goroutine as close as possible
 	// to the VM startup (below) to avoid "tart ip" timing out
 	if vm.resource.StartupScript != nil {

internal/worker/vmmanager/tart/tart.go

@@ -60,14 +60,16 @@ func NewVM(
 ) *VM {
 	vmContext, vmContextCancel := context.WithCancel(context.Background())
 
+	logger = logger.With(
+		"vm_uid", vmResource.UID,
+		"vm_name", vmResource.Name,
+		"vm_restart_count", vmResource.RestartCount,
+	)
+
 	vm := &VM{
 		onDiskName: ondiskname.NewFromResource(vmResource),
 		resource:   vmResource,
-		logger: logger.With(
-			"vm_uid", vmResource.UID,
-			"vm_name", vmResource.Name,
-			"vm_restart_count", vmResource.RestartCount,
-		),
+		logger:     logger,
 
 		ctx:    vmContext,
 		cancel: vmContextCancel,
@@ -303,6 +305,9 @@ func (vm *VM) run(ctx context.Context, eventStreamer *client.EventStreamer) {
 
 	resource := vm.Resource()
 
+	vm.EndpointSet().Start()
+	defer vm.EndpointSet().Stop()
+
 	// Launch the startup script goroutine as close as possible
 	// to the VM startup (below) to avoid "tart ip" timing out
 	if resource.StartupScript != nil {
@@ -421,6 +426,7 @@ func (vm *VM) IP(ctx context.Context) (string, error) {
 }
 
 func (vm *VM) Suspend() <-chan error {
+	vm.EndpointSet().Stop()
 	errCh := make(chan error, 1)
 
 	select {
@@ -453,6 +459,7 @@ func (vm *VM) Suspend() <-chan error {
 }
 
 func (vm *VM) Stop() <-chan error {
+	vm.EndpointSet().Stop()
 	vm.stopMtx.Lock()
 	defer vm.stopMtx.Unlock()
 	if vm.stopDone != nil {

internal/worker/vmmanager/vmmanager.go

@@ -3,6 +3,7 @@ package vmmanager
 import (
 	"context"
 
+	"github.com/cirruslabs/orchard/internal/worker/endpoint"
 	"github.com/cirruslabs/orchard/internal/worker/ondiskname"
 	"github.com/cirruslabs/orchard/pkg/client"
 	v1 "github.com/cirruslabs/orchard/pkg/resource/v1"
@@ -15,6 +16,7 @@ type VM interface {
 	UpdateSoftnetPolicy(ctx context.Context, allow []string, block []string) error
 	OnDiskName() ondiskname.OnDiskName
 	ImageFQN() *string
+	EndpointSet() *endpoint.Set
 	Status() v1.VMStatus
 	StatusMessage() string
 	Err() error

internal/worker/worker.go

@@ -14,6 +14,7 @@ import (
 	"github.com/cirruslabs/orchard/internal/dialer"
 	"github.com/cirruslabs/orchard/internal/opentelemetry"
 	"github.com/cirruslabs/orchard/internal/worker/dhcpleasetime"
+	"github.com/cirruslabs/orchard/internal/worker/endpoint"
 	"github.com/cirruslabs/orchard/internal/worker/ondiskname"
 	"github.com/cirruslabs/orchard/internal/worker/platform"
 	"github.com/cirruslabs/orchard/internal/worker/runtime"
@@ -447,6 +448,8 @@ func (worker *Worker) registerWorker(ctx context.Context) error {
 		MachineID:     platformUUID,
 		DefaultCPU:    worker.defaultCPU,
 		DefaultMemory: worker.defaultMemory,
+		Capabilities: lo.Ternary(worker.runtime.ID() == v1.RuntimeTart || worker.runtime.Synthetic(),
+			v1.WorkerCapabilities{v1.WorkerCapabilityVMEndpoints}, nil),
 	})
 	if err != nil {
 		return err
@@ -603,7 +606,7 @@ func (worker *Worker) syncVMs(
 					currentVMResource.NetSoftnetBlock = vmResource.NetSoftnetBlock
 
 					// Advance the generation only if no other spec changes remain
-					if currentVMResource.SemanticallyEqual(vmResource.VMSpec) {
+					if v1.SemanticallyEqual(currentVMResource.VMSpec, vmResource.VMSpec) {
 						currentVMResource = *vmResource
 					}
 
@@ -642,6 +645,7 @@ func (worker *Worker) syncVMs(
 
 			vmResource.Status = v1.VMStatusFailed
 			vmResource.StatusMessage = statusMessage
+			vmResource.ObservedEndpoints = nil
 			if err := updateVM(ctx, *vmResource); err != nil {
 				return err
 			}
@@ -665,10 +669,45 @@ func (worker *Worker) monitorRunningVM(
 	vm vmmanager.VM,
 	updateVM func(context.Context, v1.VM) error,
 ) error {
+	currentVMResource := vm.Resource()
+
+	// Tracks whether any specification changes were applied without restarting the VM
+	appliedInPlace := false
+
+	// Endpoint specification changes do not require restarting the VM;
+	// reconciliation happens separately below
+	endpointsChanged := !v1.SemanticallyEqual(
+		currentVMResource.Endpoints,
+		vmResource.Endpoints,
+	)
+
+	if vmResource.PowerState == v1.PowerStateRunning &&
+		!v1.ConditionIsTrue(vm.Conditions(), v1.ConditionTypeStopping) &&
+		!v1.ConditionIsTrue(vm.Conditions(), v1.ConditionTypeSuspending) &&
+		v1.ConditionIsTrue(vm.Conditions(), v1.ConditionTypeRunning) &&
+		endpointsChanged {
+		currentVMResource.Endpoints = vmResource.Endpoints
+		appliedInPlace = true
+	}
+
+	// Advance the generation only if no other specification changes remain
+	if appliedInPlace {
+		if v1.SemanticallyEqual(currentVMResource.VMSpec, vmResource.VMSpec) {
+			currentVMResource = *vmResource
+		}
+
+		vm.SetResource(currentVMResource)
+	}
+
 	worker.reconcileRunningVM(vmResource, vm) //nolint:contextcheck // Event streams outlive sync sessions.
 
 	var updateNeeded bool
 
+	if (worker.runtime.ID() == v1.RuntimeTart || worker.runtime.Synthetic()) &&
+		worker.reconcileEndpoints(vmResource, vm) {
+		updateNeeded = true
+	}
+
 	if vmResource.StatusMessage != vm.StatusMessage() {
 		vmResource.StatusMessage = vm.StatusMessage()
 
@@ -695,6 +734,34 @@ func (worker *Worker) monitorRunningVM(
 	return nil
 }
 
+func (worker *Worker) reconcileEndpoints(
+	vmResource *v1.VM,
+	vm vmmanager.VM,
+) bool {
+	// Expose endpoints only when the requested VM generation is running
+	endpointsShouldRun := vmResource.PowerState == v1.PowerStateRunning &&
+		v1.ConditionIsTrue(vm.Conditions(), v1.ConditionTypeRunning) &&
+		vmResource.Generation == vm.Resource().Generation
+
+	desired := vmResource.Endpoints
+	if !endpointsShouldRun {
+		desired = nil
+	}
+
+	observed := vm.EndpointSet().Reconcile(
+		desired,
+		endpoint.NewVMTargetBinder(vm.IP, worker.dialer),
+	)
+
+	if slices.Equal(vmResource.ObservedEndpoints, observed) {
+		return false
+	}
+
+	vmResource.ObservedEndpoints = observed
+
+	return true
+}
+
 func (worker *Worker) reconcileRunningVM(vmResource *v1.VM, vm vmmanager.VM) {
 	if vmResource.Generation == vm.Resource().Generation {
 		return

internal/worker/worker_stop_test.go

@@ -10,7 +10,10 @@ import (
 	"time"
 
 	"github.com/cirruslabs/orchard/internal/worker/ondiskname"
+	"github.com/cirruslabs/orchard/internal/worker/runtime"
 	"github.com/cirruslabs/orchard/internal/worker/vmmanager"
+	"github.com/cirruslabs/orchard/internal/worker/vmmanager/base"
+	"github.com/cirruslabs/orchard/internal/worker/vmmanager/synthetic"
 	"github.com/cirruslabs/orchard/pkg/client"
 	v1 "github.com/cirruslabs/orchard/pkg/resource/v1"
 	"github.com/stretchr/testify/require"
@@ -82,8 +85,9 @@ func TestMonitorWaitsForStopBeforeApplyingGeneration(t *testing.T) {
 			defer server.Close()
 			apiClient, err := client.New(client.WithAddress(server.URL))
 			require.NoError(t, err)
-			worker := &Worker{client: apiClient}
+			worker := &Worker{client: apiClient, runtime: runtime.NewSynthetic()}
 			vm := &delayedStopVM{
+				VM:          &synthetic.VM{VM: base.NewVM(zap.NewNop().Sugar())},
 				resource:    v1.VM{Meta: v1.Meta{Name: "test-vm"}},
 				stopStarted: make(chan struct{}),
 				stopResult:  make(chan error),

pkg/resource/v1/cmp_test.go

@@ -14,11 +14,11 @@ func TestVM(t *testing.T) {
 	cmp.Equal(v1.VM{}, v1.VM{})
 }
 
-func TestVMSpecSemanticallyEqualEquatesEmptySlices(t *testing.T) {
+func TestSemanticallyEqualEquatesEmptySlices(t *testing.T) {
 	nilSlicesSpec := v1.VMSpec{}
 	emptySlicesSpec := v1.VMSpec{
 		NetSoftnetAllow: []string{},
 		NetSoftnetBlock: []string{},
 	}
-	require.True(t, nilSlicesSpec.SemanticallyEqual(emptySlicesSpec))
+	require.True(t, v1.SemanticallyEqual(nilSlicesSpec, emptySlicesSpec))
 }

pkg/resource/v1/connection_target.go

@@ -0,0 +1,20 @@
+package v1
+
+import "fmt"
+
+type ConnectionTarget struct {
+	VM *ConnectionTargetVM `json:"vm,omitempty"`
+}
+
+type ConnectionTargetVM struct {
+	Port uint16 `json:"port"`
+}
+
+//nolint:err113,perfsprint // preserve validation errors exposed to API clients
+func (target ConnectionTarget) Validate() error {
+	if target.VM == nil || target.VM.Port == 0 {
+		return fmt.Errorf("a VM connection target with a non-zero port is required")
+	}
+
+	return nil
+}

pkg/resource/v1/endpoint.go

@@ -0,0 +1,80 @@
+//nolint:err113,perfsprint // preserve the original endpoint validation errors
+package v1
+
+import "fmt"
+
+type EndpointSpec struct {
+	Name            string           `json:"name"`
+	Target          ConnectionTarget `json:"target"`
+	WorkerPortRange *PortRange       `json:"workerPortRange,omitempty"`
+}
+
+func (endpoint EndpointSpec) Validate() error {
+	if endpoint.Name == "" {
+		return fmt.Errorf("endpoint name cannot be empty")
+	}
+
+	if err := endpoint.Target.Validate(); err != nil {
+		return fmt.Errorf("endpoint %q: %w", endpoint.Name, err)
+	}
+
+	if portRange := endpoint.WorkerPortRange; portRange != nil {
+		if err := portRange.Validate(); err != nil {
+			return fmt.Errorf("endpoint %q has invalid worker port range: %w", endpoint.Name, err)
+		}
+	}
+
+	return nil
+}
+
+type EndpointStatus struct {
+	Name       string        `json:"name"`
+	WorkerPort uint16        `json:"workerPort,omitempty"`
+	State      EndpointState `json:"state"`
+	Message    string        `json:"message,omitempty"`
+}
+
+type PortRange struct {
+	Min uint16 `json:"min"`
+	Max uint16 `json:"max"`
+}
+
+func (portRange PortRange) Validate() error {
+	if portRange.Min == 0 {
+		return fmt.Errorf("minimum port must be greater than zero")
+	}
+
+	if portRange.Min > portRange.Max {
+		return fmt.Errorf(
+			"minimum port %d exceeds maximum port %d",
+			portRange.Min,
+			portRange.Max,
+		)
+	}
+
+	return nil
+}
+
+type EndpointState string
+
+const (
+	EndpointStateListening EndpointState = "listening"
+	EndpointStateError     EndpointState = "error"
+)
+
+func ValidateEndpoints(endpoints []EndpointSpec) error {
+	seenNames := make(map[string]struct{}, len(endpoints))
+
+	for _, endpoint := range endpoints {
+		if err := endpoint.Validate(); err != nil {
+			return err
+		}
+		if _, exists := seenNames[endpoint.Name]; exists {
+			return fmt.Errorf("endpoint %q is duplicated", endpoint.Name)
+		}
+
+		seenNames[endpoint.Name] = struct{}{}
+	}
+
+	return nil
+}

pkg/resource/v1/v1.go

@@ -189,17 +189,18 @@ type VMSpec struct {
 	// so this field defaults to that when not set.
 	Runtime Runtime `json:"runtime,omitempty"`
 
-	NetSoftnetDeprecated bool       `json:"net-softnet,omitempty"`
-	NetSoftnet           bool       `json:"netSoftnet,omitempty"`
-	NetSoftnetAllow      []string   `json:"netSoftnetAllow,omitempty"`
-	NetSoftnetBlock      []string   `json:"netSoftnetBlock,omitempty"`
-	Suspendable          bool       `json:"suspendable,omitempty"`
-	PowerState           PowerState `json:"powerState,omitempty"`
+	Endpoints            []EndpointSpec `json:"endpoints,omitempty"`
+	NetSoftnetDeprecated bool           `json:"net-softnet,omitempty"` //nolint:tagliatelle // legacy JSON key
+	NetSoftnet           bool           `json:"netSoftnet,omitempty"`
+	NetSoftnetAllow      []string       `json:"netSoftnetAllow,omitempty"`
+	NetSoftnetBlock      []string       `json:"netSoftnetBlock,omitempty"`
+	Suspendable          bool           `json:"suspendable,omitempty"`
+	PowerState           PowerState     `json:"powerState,omitempty"`
 }
 
-func (vm VMSpec) SemanticallyEqual(other VMSpec) bool {
-	// Treat omitted and explicitly empty collections as the same VM specification
-	return cmp.Equal(vm, other, cmpopts.EquateEmpty())
+// SemanticallyEqual treats omitted and explicitly empty collections as equal.
+func SemanticallyEqual[T any](current, desired T) bool {
+	return cmp.Equal(current, desired, cmpopts.EquateEmpty())
 }
 
 func (vm VMSpec) SoftnetEnabled() bool {
@@ -224,6 +225,9 @@ type VMState struct {
 	// on which the worker had acted upon.
 	ObservedGeneration uint64 `json:"observedGeneration"`
 
+	// ObservedEndpoints contains the current endpoint statuses.
+	ObservedEndpoints []EndpointStatus `json:"observedEndpoints,omitempty"`
+
 	Conditions []Condition `json:"conditions,omitempty"`
 }
 

pkg/resource/v1/worker.go

@@ -30,6 +30,8 @@ type Worker struct {
 	// Runtime defines a runtime provided by this worker.
 	Runtime Runtime `json:"runtime,omitempty"`
 
+	Capabilities WorkerCapabilities `json:"capabilities,omitempty"`
+
 	Meta
 }
 
@@ -39,6 +41,23 @@ func (worker Worker) Offline(workerOfflineTimeout time.Duration) bool {
 
 func (worker *Worker) SetVersion(_ uint64) {}
 
+type WorkerCapability string
+
+const WorkerCapabilityVMEndpoints WorkerCapability = "vm-endpoints"
+
+type WorkerCapabilities []WorkerCapability
+
+//nolint:modernize // preserve the original capability membership helper
+func (workerCapabilities WorkerCapabilities) Has(capability WorkerCapability) bool {
+	for _, workerCapability := range workerCapabilities {
+		if workerCapability == capability {
+			return true
+		}
+	}
+
+	return false
+}
+
 func (worker *Worker) Match(filter Filter) bool {
 	return false
 }