feat: optimize shared desktop access and multi-replica control plane

- Add leader election to avoid duplicate sync and Team event consumers
- Proxy desktop traffic directly from nginx to instance Services to reduce load during concurrent viewers
- Add desktop stream profiles with low / standard / high options
- Limit hostPath PV fallback to manual storageClass only

Dockerfile

@@ -30,13 +30,16 @@ RUN CGO_ENABLED=0 GOOS=${TARGETOS:-linux} GOARCH=${TARGETARCH} go build -trimpat
 
 FROM nginx:1.27-alpine
 
-RUN apk add --no-cache dumb-init openssl
+# nginx-module-njs provides ngx_http_js_module.so, loaded by nginx.conf to run
+# the desktop access-token verification (deployments/nginx/njs/desktop_auth.js).
+RUN apk add --no-cache dumb-init openssl nginx-module-njs
 
 WORKDIR /app
 
 COPY --from=backend-builder /out/clawreef-server /usr/local/bin/clawreef-server
 COPY --from=frontend-builder /app/frontend/dist /usr/share/nginx/html
 COPY deployments/nginx/nginx.conf /etc/nginx/nginx.conf
+COPY deployments/nginx/njs/desktop_auth.js /etc/nginx/njs/desktop_auth.js
 COPY deployments/container/start.sh /app/start.sh
 
 RUN chmod +x /app/start.sh \

backend/cmd/server/main.go

@@ -7,6 +7,7 @@ import (
 	"os"
 	"os/signal"
 	"strings"
+	"sync"
 	"syscall"
 	"time"
 
@@ -19,6 +20,7 @@ import (
 	"clawreef/internal/repository"
 	"clawreef/internal/services"
 	"clawreef/internal/services/k8s"
+	"clawreef/internal/services/leader"
 
 	"github.com/gin-gonic/gin"
 )
@@ -177,18 +179,21 @@ func main() {
 		services.StartRuntimeAdminEventBridge(bridgeCtx, runtimeEvents, wsHub)
 	}
 
-	// Start sync service to keep instance status in sync with K8s
+	// Control-plane singleton background loops. The HTTP API and the in-pod
+	// nginx desktop data plane run on every replica, but these loops must run
+	// on exactly one replica. With leader election enabled they only run on the
+	// elected leader and migrate on failover; with it disabled (single-replica
+	// deployments) they run directly.
 	syncService := services.NewSyncService(instanceRepo, instanceRuntimeStatusService)
-	syncService.Start()
-	teamService.Start()
 	var runtimeSchedulerCancel context.CancelFunc
+	var runtimeSchedulerMu sync.Mutex
 	var runtimeScheduler *services.RuntimeScheduler
 	if cfg.Runtime.SchedulerEnabled {
 		k8sClient := k8s.GetClient()
 		if k8sClient == nil || k8sClient.Clientset == nil {
 			log.Printf("runtime scheduler disabled: k8s client is unavailable")
 		} else {
-			leader := services.NewRuntimeLeaderService(k8sClient.Clientset, cfg.Runtime.Namespace, cfg.Runtime.BackendReplicaID)
+			runtimeLeader := services.NewRuntimeLeaderService(k8sClient.Clientset, cfg.Runtime.Namespace, cfg.Runtime.BackendReplicaID)
 			runtimeDeployments := k8s.NewRuntimeDeploymentService(k8sClient.Clientset)
 			runtimeSchedulerOptions := []services.RuntimeSchedulerOption{
 				services.WithRuntimeSchedulerWorkspaceRoot(cfg.Runtime.WorkspaceRoot),
@@ -209,21 +214,65 @@ func main() {
 				rolloutRepo,
 				runtimeAgentClient,
 				runtimeEvents,
-				leader,
+				runtimeLeader,
 				runtimeDeployments,
 				cfg.Runtime.SchedulerTick,
 				runtimeSchedulerOptions...,
 			)
-			var schedulerCtx context.Context
-			schedulerCtx, runtimeSchedulerCancel = context.WithCancel(context.Background())
-			runtimeScheduler.Start(schedulerCtx)
-			log.Printf("runtime scheduler started")
+			log.Printf("runtime scheduler initialized")
 		}
 	} else {
 		log.Printf("runtime scheduler disabled by configuration")
 	}
 	runtimePoolHandler := handlers.NewRuntimePoolHandler(runtimePodRepo, bindingRepo, rolloutRepo, runtimeScheduler, runtimeEvents)
 
+	leaderCtx, leaderCancel := context.WithCancel(context.Background())
+	defer leaderCancel()
+
+	startBackground := func(ctx context.Context) {
+		log.Printf("Starting leader-only background loops (identity=%s)", cfg.LeaderElection.Identity)
+		syncService.Start()
+		teamService.StartBackground(ctx)
+		if runtimeScheduler != nil {
+			runtimeSchedulerMu.Lock()
+			if runtimeSchedulerCancel == nil {
+				var schedulerCtx context.Context
+				schedulerCtx, runtimeSchedulerCancel = context.WithCancel(ctx)
+				runtimeScheduler.Start(schedulerCtx)
+				log.Printf("runtime scheduler started")
+			}
+			runtimeSchedulerMu.Unlock()
+		}
+	}
+	stopBackground := func() {
+		log.Printf("Stopping leader-only background loops (identity=%s)", cfg.LeaderElection.Identity)
+		runtimeSchedulerMu.Lock()
+		if runtimeSchedulerCancel != nil {
+			runtimeSchedulerCancel()
+			runtimeSchedulerCancel = nil
+		}
+		runtimeSchedulerMu.Unlock()
+		teamService.StopBackground()
+		syncService.Stop()
+	}
+
+	if cfg.LeaderElection.Enabled && k8s.GetClient() != nil && k8s.GetClient().Clientset != nil {
+		go leader.Run(leaderCtx, k8s.GetClient().Clientset, leader.Config{
+			Namespace:     cfg.LeaderElection.Namespace,
+			LeaseName:     cfg.LeaderElection.LeaseName,
+			Identity:      cfg.LeaderElection.Identity,
+			LeaseDuration: time.Duration(cfg.LeaderElection.LeaseDuration) * time.Second,
+			RenewDeadline: time.Duration(cfg.LeaderElection.RenewDeadline) * time.Second,
+			RetryPeriod:   time.Duration(cfg.LeaderElection.RetryPeriod) * time.Second,
+		}, leader.Callbacks{
+			OnStartedLeading: startBackground,
+			OnStoppedLeading: stopBackground,
+		})
+	} else {
+		log.Println("Leader election disabled or K8s unavailable; running control-plane background loops directly")
+		startBackground(leaderCtx)
+	}
+
 	// Setup router
 	r := gin.Default()
 
@@ -540,15 +589,14 @@ func main() {
 		log.Printf("HTTP server forced to shutdown: %v", err)
 	}
 
-	// Stop background services
-	if runtimeSchedulerCancel != nil {
-		runtimeSchedulerCancel()
-	}
+	// Stop background services. Cancelling leaderCtx releases the lease (and,
+	// if we were leader, triggers stopBackground); the explicit stopBackground
+	// call is idempotent and covers the leader-election-disabled path.
+	leaderCancel()
+	stopBackground()
 	if runtimeAdminEventBridgeCancel != nil {
 		runtimeAdminEventBridgeCancel()
 	}
-	syncService.Stop()
-	teamService.Stop()
 	wsHub.Stop()
 	instanceHandler.Shutdown()
 

backend/internal/config/config.go

@@ -12,13 +12,29 @@ import (
 
 // Config holds all application configuration
 type Config struct {
-	Server        ServerConfig        `yaml:"server"`
-	Database      DatabaseConfig      `yaml:"database"`
-	JWT           JWTConfig           `yaml:"jwt"`
-	Kubernetes    KubernetesConfig    `yaml:"kubernetes"`
-	Runtime       RuntimePoolConfig   `yaml:"runtime"`
-	ObjectStorage ObjectStorageConfig `yaml:"objectStorage"`
-	SkillScanner  SkillScannerConfig  `yaml:"skillScanner"`
+	Server         ServerConfig         `yaml:"server"`
+	Database       DatabaseConfig       `yaml:"database"`
+	JWT            JWTConfig            `yaml:"jwt"`
+	Kubernetes     KubernetesConfig     `yaml:"kubernetes"`
+	Runtime        RuntimePoolConfig    `yaml:"runtime"`
+	ObjectStorage  ObjectStorageConfig  `yaml:"objectStorage"`
+	SkillScanner   SkillScannerConfig   `yaml:"skillScanner"`
+	LeaderElection LeaderElectionConfig `yaml:"leaderElection"`
+}
+
+// LeaderElectionConfig controls the control-plane leader election that gates
+// singleton background loops (SyncService + Team event consumers + stale-task
+// monitor) so they run on exactly one replica when clawmanager-app is scaled
+// horizontally. The HTTP API and the in-pod nginx data plane run on every
+// replica regardless.
+type LeaderElectionConfig struct {
+	Enabled       bool   `yaml:"enabled"`
+	Namespace     string `yaml:"namespace"`
+	LeaseName     string `yaml:"leaseName"`
+	Identity      string `yaml:"identity"`
+	LeaseDuration int    `yaml:"leaseDuration"` // seconds
+	RenewDeadline int    `yaml:"renewDeadline"` // seconds
+	RetryPeriod   int    `yaml:"retryPeriod"`   // seconds
 }
 
 // ServerConfig holds server-related configuration
@@ -258,6 +274,15 @@ func Load() (*Config, error) {
 			TimeoutSeconds: 30,
 			Enabled:        strings.EqualFold(getEnv("SKILL_SCANNER_ENABLED", "false"), "true"),
 		},
+		LeaderElection: LeaderElectionConfig{
+			Enabled:       strings.EqualFold(getEnv("CLAWMANAGER_LEADER_ELECTION", "true"), "true"),
+			Namespace:     getEnv("POD_NAMESPACE", "clawmanager-system"),
+			LeaseName:     getEnv("CLAWMANAGER_LEADER_LEASE_NAME", "clawmanager-controlplane-leader"),
+			Identity:      getEnv("POD_NAME", ""),
+			LeaseDuration: 15,
+			RenewDeadline: 10,
+			RetryPeriod:   2,
+		},
 	}
 
 	// Try to load from k8s config file
@@ -372,6 +397,13 @@ func applyEnvOverrides(config *Config) {
 	config.Runtime.MaxGatewaysPerPod = getEnvInt("RUNTIME_MAX_GATEWAYS_PER_POD", config.Runtime.MaxGatewaysPerPod)
 	config.Runtime.GatewayPortStart = getEnvInt("RUNTIME_GATEWAY_PORT_START", config.Runtime.GatewayPortStart)
 	config.Runtime.GatewayPortEnd = getEnvInt("RUNTIME_GATEWAY_PORT_END", config.Runtime.GatewayPortEnd)
+	config.LeaderElection.Enabled = getEnvBool("CLAWMANAGER_LEADER_ELECTION", config.LeaderElection.Enabled)
+	config.LeaderElection.Namespace = getEnv("POD_NAMESPACE", config.LeaderElection.Namespace)
+	config.LeaderElection.LeaseName = getEnv("CLAWMANAGER_LEADER_LEASE_NAME", config.LeaderElection.LeaseName)
+	config.LeaderElection.Identity = getEnv("POD_NAME", config.LeaderElection.Identity)
+	config.LeaderElection.LeaseDuration = getEnvInt("CLAWMANAGER_LEADER_LEASE_DURATION", config.LeaderElection.LeaseDuration)
+	config.LeaderElection.RenewDeadline = getEnvInt("CLAWMANAGER_LEADER_RENEW_DEADLINE", config.LeaderElection.RenewDeadline)
+	config.LeaderElection.RetryPeriod = getEnvInt("CLAWMANAGER_LEADER_RETRY_PERIOD", config.LeaderElection.RetryPeriod)
 
 	if endpoint := os.Getenv("OBJECT_STORAGE_ENDPOINT"); endpoint != "" {
 		config.ObjectStorage.Endpoint = endpoint

backend/internal/handlers/instance_handler.go

@@ -30,6 +30,31 @@ const (
 	workspaceArchiveMaxMiBEnv     = "CLAWMANAGER_WORKSPACE_ARCHIVE_MAX_MIB"
 )
 
+// desktopDirectProxyEnv toggles embedding the instance Service "host:port" into
+// the access token so the edge gateway can proxy desktop traffic directly to the
+// instance instead of relaying it through this control-plane process.
+const desktopDirectProxyEnv = "CLAWMANAGER_DESKTOP_DIRECT_PROXY"
+
+// desktopDirectProxyEnabled reports whether direct desktop proxying is enabled.
+func desktopDirectProxyEnabled() bool {
+	switch strings.ToLower(strings.TrimSpace(os.Getenv(desktopDirectProxyEnv))) {
+	case "1", "true", "yes", "on":
+		return true
+	default:
+		return false
+	}
+}
+
+func desktopProxyMode(directEnabled bool, upstream string) string {
+	if !directEnabled {
+		return "control-plane"
+	}
+	if strings.TrimSpace(upstream) == "" {
+		return "fallback"
+	}
+	return "direct"
+}
+
 func workspaceArchiveMaxMiB() int64 {
 	value := strings.TrimSpace(os.Getenv(workspaceArchiveMaxMiBEnv))
 	if value == "" {
@@ -112,10 +137,11 @@ type ExternalAccessRequest struct {
 type CreateInstanceRequest struct {
 	Name                 string                       `json:"name" binding:"required,min=3,max=50"`
 	Description          *string                      `json:"description,omitempty"`
-	Type                 string                       `json:"type" binding:"required,oneof=openclaw hermes"`
+	Type                 string                       `json:"type" binding:"required,oneof=openclaw ubuntu debian centos custom webtop hermes"`
 	Mode                 string                       `json:"mode" binding:"omitempty,oneof=lite pro"`
 	InstanceMode         string                       `json:"instance_mode" binding:"omitempty,oneof=lite pro"`
 	RuntimeType          string                       `json:"runtime_type" binding:"omitempty,oneof=gateway desktop shell"`
+	DesktopStreamProfile string                       `json:"desktop_stream_profile,omitempty" binding:"omitempty,oneof=low standard high"`
 	CPUCores             float64                      `json:"cpu_cores" binding:"required,min=0.1,max=32"`
 	MemoryGB             int                          `json:"memory_gb" binding:"required,min=1,max=128"`
 	DiskGB               int                          `json:"disk_gb" binding:"required,min=10,max=1000"`
@@ -133,8 +159,9 @@ type CreateInstanceRequest struct {
 
 // UpdateInstanceRequest represents an update instance request
 type UpdateInstanceRequest struct {
-	Name        *string `json:"name,omitempty" binding:"omitempty,min=3,max=50"`
-	Description *string `json:"description,omitempty"`
+	Name                 *string `json:"name,omitempty" binding:"omitempty,min=3,max=50"`
+	Description          *string `json:"description,omitempty"`
+	DesktopStreamProfile *string `json:"desktop_stream_profile,omitempty" binding:"omitempty,oneof=low standard high"`
 }
 
 // ListInstancesRequest represents a list instances request
@@ -225,6 +252,7 @@ func (h *InstanceHandler) CreateInstance(c *gin.Context) {
 		Mode:                 req.Mode,
 		InstanceMode:         req.InstanceMode,
 		RuntimeType:          req.RuntimeType,
+		DesktopStreamProfile: req.DesktopStreamProfile,
 		CPUCores:             req.CPUCores,
 		MemoryGB:             req.MemoryGB,
 		DiskGB:               req.DiskGB,
@@ -369,8 +397,9 @@ func (h *InstanceHandler) UpdateInstance(c *gin.Context) {
 	}
 
 	updateReq := services.UpdateInstanceRequest{
-		Name:        req.Name,
-		Description: req.Description,
+		Name:                 req.Name,
+		Description:          req.Description,
+		DesktopStreamProfile: req.DesktopStreamProfile,
 	}
 
 	if err := h.instanceService.Update(id, updateReq); err != nil {
@@ -845,6 +874,39 @@ func (h *InstanceHandler) GenerateAccessToken(c *gin.Context) {
 		return
 	}
 
+	targetPort := h.proxyService.GetTargetPortForInstance(instance)
+
+	// When direct desktop proxying is enabled, embed the instance Service
+	// "host:port" into the token so the edge gateway can dial the instance
+	// directly. On any failure we fall back to an empty upstream, which keeps
+	// the request flowing through the in-process control-plane proxy.
+	upstream := ""
+	directProxyEnabled := desktopDirectProxyEnabled()
+	if directProxyEnabled {
+		if h.proxyService.IsWebtopInstanceType(instance.Type) {
+			resolved, resolveErr := h.proxyService.ResolveUpstreamHostPort(
+				c.Request.Context(),
+				instance.UserID,
+				instance.ID,
+				targetPort,
+			)
+			if resolveErr != nil {
+				fmt.Printf("Desktop direct proxy fallback: failed to resolve upstream instance=%d user=%d type=%s target_port=%d error=%v\n",
+					instance.ID, instance.UserID, instance.Type, targetPort, resolveErr)
+			} else if strings.TrimSpace(resolved) == "" {
+				fmt.Printf("Desktop direct proxy fallback: resolved empty upstream instance=%d user=%d type=%s target_port=%d\n",
+					instance.ID, instance.UserID, instance.Type, targetPort)
+			} else {
+				upstream = resolved
+				fmt.Printf("Desktop direct proxy resolved: instance=%d user=%d type=%s target_port=%d upstream=%s\n",
+					instance.ID, instance.UserID, instance.Type, targetPort, upstream)
+			}
+		} else {
+			fmt.Printf("Desktop direct proxy fallback: unsupported desktop instance type instance=%d user=%d type=%s target_port=%d\n",
+				instance.ID, instance.UserID, instance.Type, targetPort)
+		}
+	}
+
 	// Generate access token (valid for 1 hour)
 	maxAgeSeconds := int(time.Hour.Seconds())
 	token, err := h.accessService.GenerateToken(
@@ -852,7 +914,8 @@ func (h *InstanceHandler) GenerateAccessToken(c *gin.Context) {
 		instance.ID,
 		instance.Type,
 		accessURL,
-		h.proxyService.GetTargetPortForInstance(instance),
+		upstream,
+		targetPort,
 		1*time.Hour,
 	)
 	if err != nil {
@@ -874,10 +937,12 @@ func (h *InstanceHandler) GenerateAccessToken(c *gin.Context) {
 
 	// Return token and URLs
 	response := map[string]interface{}{
-		"token":      token.Token,
-		"access_url": accessURL,
-		"proxy_url":  h.proxyService.GetProxyURLForInstance(instance, token.Token),
-		"expires_at": token.ExpiresAt,
+		"token":                    token.Token,
+		"access_url":               accessURL,
+		"proxy_url":                h.proxyService.GetProxyURLForInstance(instance, token.Token),
+		"expires_at":               token.ExpiresAt,
+		"desktop_proxy_mode":       desktopProxyMode(directProxyEnabled, upstream),
+		"desktop_upstream_present": upstream != "",
 	}
 
 	utils.Success(c, http.StatusOK, "Access token generated successfully", response)

backend/internal/models/instance.go

@@ -25,6 +25,7 @@ type Instance struct {
 	ImageRegistry            *string    `db:"image_registry" json:"image_registry,omitempty"`
 	ImageTag                 *string    `db:"image_tag" json:"image_tag,omitempty"`
 	EnvironmentOverridesJSON *string    `db:"environment_overrides_json" json:"-"`
+	DesktopStreamProfile     string     `db:"-" json:"desktop_stream_profile,omitempty"`
 	StorageClass             string     `db:"storage_class" json:"storage_class"`
 	MountPath                string     `db:"mount_path" json:"mount_path"`
 	WorkspacePath            *string    `db:"workspace_path" json:"workspace_path,omitempty"`

backend/internal/services/desktop_stream_profile.go

@@ -0,0 +1,90 @@
+package services
+
+import "strings"
+
+const (
+	DesktopStreamProfileLow      = "low"
+	DesktopStreamProfileStandard = "standard"
+	DesktopStreamProfileHigh     = "high"
+
+	desktopStreamProfileEnvKey = "CLAWMANAGER_DESKTOP_STREAM_PROFILE"
+)
+
+var selkiesDesktopStreamEnvKeys = []string{
+	desktopStreamProfileEnvKey,
+	"SELKIES_ENCODER",
+	"SELKIES_FRAMERATE",
+	"SELKIES_H264_CRF",
+	"SELKIES_AUDIO_ENABLED",
+	"SELKIES_SECOND_SCREEN",
+}
+
+func normalizeDesktopStreamProfile(profile string) (string, bool) {
+	switch strings.ToLower(strings.TrimSpace(profile)) {
+	case "":
+		return "", true
+	case DesktopStreamProfileLow:
+		return DesktopStreamProfileLow, true
+	case DesktopStreamProfileStandard:
+		return DesktopStreamProfileStandard, true
+	case DesktopStreamProfileHigh:
+		return DesktopStreamProfileHigh, true
+	default:
+		return "", false
+	}
+}
+
+func applyDesktopStreamProfileEnv(overrides map[string]string, profile string) map[string]string {
+	normalized, ok := normalizeDesktopStreamProfile(profile)
+	if !ok || normalized == "" {
+		return overrides
+	}
+
+	if overrides == nil {
+		overrides = map[string]string{}
+	}
+
+	for _, key := range selkiesDesktopStreamEnvKeys {
+		delete(overrides, key)
+	}
+
+	overrides[desktopStreamProfileEnvKey] = normalized
+	overrides["SELKIES_ENCODER"] = "x264enc"
+	overrides["SELKIES_SECOND_SCREEN"] = "false"
+	overrides["SELKIES_AUDIO_ENABLED"] = "false"
+
+	switch normalized {
+	case DesktopStreamProfileLow:
+		overrides["SELKIES_FRAMERATE"] = "30"
+		overrides["SELKIES_H264_CRF"] = "42"
+	case DesktopStreamProfileHigh:
+		overrides["SELKIES_FRAMERATE"] = "40"
+		overrides["SELKIES_H264_CRF"] = "24"
+	default:
+		overrides["SELKIES_FRAMERATE"] = "35"
+		overrides["SELKIES_H264_CRF"] = "34"
+	}
+
+	return overrides
+}
+
+func desktopStreamProfileFromEnv(overrides map[string]string) string {
+	if profile, ok := overrides[desktopStreamProfileEnvKey]; ok {
+		if normalized, valid := normalizeDesktopStreamProfile(profile); valid && normalized != "" {
+			return normalized
+		}
+	}
+
+	framerate := strings.TrimSpace(overrides["SELKIES_FRAMERATE"])
+	crf := strings.TrimSpace(overrides["SELKIES_H264_CRF"])
+	switch {
+	case framerate == "30" && crf == "42":
+		return DesktopStreamProfileLow
+	case framerate == "40" && crf == "24":
+		return DesktopStreamProfileHigh
+	case framerate == "35" && crf == "34":
+		return DesktopStreamProfileStandard
+	default:
+		return ""
+	}
+}

backend/internal/services/instance_access_service.go

@@ -18,6 +18,7 @@ type AccessToken struct {
 	InstanceType string    `json:"instance_type"`
 	TargetPort   int32     `json:"target_port"`
 	AccessURL    string    `json:"access_url"`
+	Upstream     string    `json:"upstream"`
 	ExpiresAt    time.Time `json:"expires_at"`
 	CreatedAt    time.Time `json:"created_at"`
 }
@@ -36,6 +37,7 @@ type instanceAccessClaims struct {
 	InstanceType string `json:"instance_type"`
 	TargetPort   int32  `json:"target_port"`
 	AccessURL    string `json:"access_url"`
+	Upstream     string `json:"upstream"`
 	TokenType    string `json:"token_type"`
 	jwt.RegisteredClaims
 }
@@ -55,7 +57,7 @@ func NewInstanceAccessService() *InstanceAccessService {
 }
 
 // GenerateToken generates a new access token for an instance
-func (s *InstanceAccessService) GenerateToken(userID, instanceID int, instanceType string, accessURL string, targetPort int32, duration time.Duration) (*AccessToken, error) {
+func (s *InstanceAccessService) GenerateToken(userID, instanceID int, instanceType string, accessURL string, upstream string, targetPort int32, duration time.Duration) (*AccessToken, error) {
 	now := time.Now()
 	expiresAt := now.Add(duration)
 
@@ -65,6 +67,7 @@ func (s *InstanceAccessService) GenerateToken(userID, instanceID int, instanceTy
 		InstanceType: instanceType,
 		TargetPort:   targetPort,
 		AccessURL:    accessURL,
+		Upstream:     upstream,
 		TokenType:    "instance_access",
 		RegisteredClaims: jwt.RegisteredClaims{
 			IssuedAt:  jwt.NewNumericDate(now),
@@ -85,6 +88,7 @@ func (s *InstanceAccessService) GenerateToken(userID, instanceID int, instanceTy
 		InstanceType: instanceType,
 		TargetPort:   targetPort,
 		AccessURL:    accessURL,
+		Upstream:     upstream,
 		ExpiresAt:    expiresAt,
 		CreatedAt:    now,
 	}
@@ -146,6 +150,7 @@ func (s *InstanceAccessService) validateSignedToken(token string) (*AccessToken,
 		InstanceType: claims.InstanceType,
 		TargetPort:   claims.TargetPort,
 		AccessURL:    claims.AccessURL,
+		Upstream:     claims.Upstream,
 		ExpiresAt:    expiresAt,
 		CreatedAt:    createdAt,
 	}, nil

backend/internal/services/instance_access_service_test.go

@@ -11,7 +11,8 @@ func TestInstanceAccessServiceValidatesTokenAcrossServiceInstances(t *testing.T)
 	issuer := NewInstanceAccessService()
 	validator := NewInstanceAccessService()
 
-	token, err := issuer.GenerateToken(7, 42, "openclaw", "/api/v1/instances/42/proxy/", 3001, 5*time.Minute)
+	wantUpstream := "clawreef-42-demo-svc.clawmanager-user-7.svc.cluster.local:3001"
+	token, err := issuer.GenerateToken(7, 42, "openclaw", "/api/v1/instances/42/proxy/", wantUpstream, 3001, 5*time.Minute)
 	if err != nil {
 		t.Fatalf("GenerateToken() error = %v", err)
 	}
@@ -30,13 +31,16 @@ func TestInstanceAccessServiceValidatesTokenAcrossServiceInstances(t *testing.T)
 	if validated.InstanceType != "openclaw" {
 		t.Fatalf("validated.InstanceType = %q, want openclaw", validated.InstanceType)
 	}
+	if validated.Upstream != wantUpstream {
+		t.Fatalf("validated.Upstream = %q, want %q", validated.Upstream, wantUpstream)
+	}
 }
 
 func TestInstanceAccessServiceRejectsExpiredSignedToken(t *testing.T) {
 	t.Setenv("INSTANCE_ACCESS_TOKEN_SECRET", "cluster-shared-secret")
 
 	service := NewInstanceAccessService()
-	token, err := service.GenerateToken(7, 42, "openclaw", "/api/v1/instances/42/proxy/", 3001, -time.Second)
+	token, err := service.GenerateToken(7, 42, "openclaw", "/api/v1/instances/42/proxy/", "", 3001, -time.Second)
 	if err != nil {
 		t.Fatalf("GenerateToken() error = %v", err)
 	}
@@ -81,7 +85,7 @@ func TestInstanceAccessServiceStopTerminatesCleanup(t *testing.T) {
 
 	// After Stop, the service should still be usable for token operations
 	// (only the background cleanup goroutine is stopped).
-	token, err := service.GenerateToken(1, 1, "openclaw", "/proxy", 3001, time.Minute)
+	token, err := service.GenerateToken(1, 1, "openclaw", "/proxy", "", 3001, time.Minute)
 	if err != nil {
 		t.Fatalf("GenerateToken after Stop() error = %v", err)
 	}

backend/internal/services/instance_proxy_service.go

@@ -659,6 +659,18 @@ func (s *InstanceProxyService) GetTargetPortForInstance(instance *models.Instanc
 	return buildRuntimeConfig(instance.Type, instance.OSType, instance.OSVersion, instance.ImageRegistry, instance.ImageTag).Port
 }
 
+// ResolveUpstreamHostPort ensures the instance Service exists and returns its
+// cluster-internal "host:port" target so the edge gateway can proxy directly to
+// the instance without routing pixel traffic through this control-plane process.
+func (s *InstanceProxyService) ResolveUpstreamHostPort(ctx context.Context, userID, instanceID int, targetPort int32) (string, error) {
+	serviceInfo, err := s.getOrCreateService(ctx, userID, instanceID, targetPort)
+	if err != nil {
+		return "", fmt.Errorf("failed to resolve upstream service: %w", err)
+	}
+
+	return fmt.Sprintf("%s.%s.svc.cluster.local:%d", serviceInfo.Name, serviceInfo.Namespace, serviceInfo.TargetPort), nil
+}
+
 func (s *InstanceProxyService) resolveTargetPort(instanceType string, defaultPort int32, targetPath string) int32 {
 	if usesWebtopImage(instanceType) {
 		if defaultPort == 0 {
@@ -734,6 +746,13 @@ func (s *InstanceProxyService) shouldRewriteHTMLForProxy(instanceID int, instanc
 	return s.shouldRewriteHTML(instanceType)
 }
 
+// IsWebtopInstanceType reports whether the instance type is served by a
+// Webtop/KasmVNC desktop image (and therefore eligible for direct gateway
+// proxying via SUBFOLDER-prefixed paths).
+func (s *InstanceProxyService) IsWebtopInstanceType(instanceType string) bool {
+	return usesWebtopImage(instanceType)
+}
+
 func (s *InstanceProxyService) getCachedService(key serviceCacheKey) *k8s.ServiceInfo {
 	s.cacheMu.RLock()
 	entry, ok := s.serviceCache[key]

backend/internal/services/instance_service.go

@@ -52,6 +52,9 @@ func (s *instanceService) ValidateCreateRequests(userID int, requests []CreateIn
 		if _, err := marshalEnvironmentOverrides(environmentOverrides); err != nil {
 			return err
 		}
+		if _, ok := normalizeDesktopStreamProfile(requests[idx].DesktopStreamProfile); !ok {
+			return fmt.Errorf("invalid desktop stream profile")
+		}
 	}
 
 	quota, err := s.quotaRepo.GetByUserID(userID)
@@ -136,10 +139,11 @@ func (s *instanceService) ValidateCreateRequests(userID int, requests []CreateIn
 type CreateInstanceRequest struct {
 	Name                 string              `json:"name" validate:"required,min=3,max=50"`
 	Description          *string             `json:"description,omitempty"`
-	Type                 string              `json:"type" validate:"required,oneof=openclaw hermes"`
+	Type                 string              `json:"type" validate:"required,oneof=openclaw ubuntu debian centos custom webtop hermes"`
 	Mode                 string              `json:"mode" validate:"omitempty,oneof=lite pro"`
 	InstanceMode         string              `json:"instance_mode" validate:"omitempty,oneof=lite pro"`
 	RuntimeType          string              `json:"runtime_type" validate:"omitempty,oneof=gateway desktop shell"`
+	DesktopStreamProfile string              `json:"desktop_stream_profile,omitempty" validate:"omitempty,oneof=low standard high"`
 	CPUCores             float64             `json:"cpu_cores" validate:"required,min=0.1,max=32"`
 	MemoryGB             int                 `json:"memory_gb" validate:"required,min=1,max=128"`
 	DiskGB               int                 `json:"disk_gb" validate:"required,min=10,max=1000"`
@@ -177,8 +181,9 @@ type instanceModeLimitConfig struct {
 
 // UpdateInstanceRequest holds data for updating an instance
 type UpdateInstanceRequest struct {
-	Name        *string `json:"name,omitempty" validate:"omitempty,min=3,max=50"`
-	Description *string `json:"description,omitempty"`
+	Name                 *string `json:"name,omitempty" validate:"omitempty,min=3,max=50"`
+	Description          *string `json:"description,omitempty"`
+	DesktopStreamProfile *string `json:"desktop_stream_profile,omitempty" validate:"omitempty,oneof=low standard high"`
 }
 
 // InstanceStatus holds the status of an instance
@@ -269,6 +274,11 @@ func (s *instanceService) Create(userID int, req CreateInstanceRequest) (*models
 	if err != nil {
 		return nil, err
 	}
+	if profile, ok := normalizeDesktopStreamProfile(req.DesktopStreamProfile); !ok {
+		return nil, fmt.Errorf("invalid desktop stream profile")
+	} else if profile != "" {
+		environmentOverrides = applyDesktopStreamProfileEnv(environmentOverrides, profile)
+	}
 	environmentOverridesJSON, err := marshalEnvironmentOverrides(environmentOverrides)
 	if err != nil {
 		return nil, err
@@ -666,6 +676,7 @@ func (s *instanceService) Create(userID int, req CreateInstanceRequest) (*models
 
 	// Broadcast initial creating status via WebSocket. Sync service will mark it
 	// running only after the pod becomes Ready.
+	hydrateInstanceDesktopStreamProfile(instance)
 	GetHub().BroadcastInstanceStatus(userID, instance)
 	fmt.Printf("Instance %d status broadcast complete\n", instance.ID)
 
@@ -722,7 +733,12 @@ func (s *instanceService) createV2Instance(ctx context.Context, userID int, req
 
 // GetByID gets an instance by ID
 func (s *instanceService) GetByID(id int) (*models.Instance, error) {
-	return s.instanceRepo.GetByID(id)
+	instance, err := s.instanceRepo.GetByID(id)
+	if err != nil {
+		return nil, err
+	}
+	hydrateInstanceDesktopStreamProfile(instance)
+	return instance, nil
 }
 
 // GetByUserID gets instances by user ID with pagination
@@ -731,6 +747,7 @@ func (s *instanceService) GetByUserID(userID int, offset, limit int) ([]models.I
 	if err != nil {
 		return nil, 0, err
 	}
+	hydrateInstancesDesktopStreamProfile(instances)
 
 	total, err := s.instanceRepo.CountByUserID(userID)
 	if err != nil {
@@ -745,6 +762,7 @@ func (s *instanceService) GetAllInstances(offset, limit int) ([]models.Instance,
 	if err != nil {
 		return nil, 0, err
 	}
+	hydrateInstancesDesktopStreamProfile(instances)
 
 	total, err := s.instanceRepo.CountAll()
 	if err != nil {
@@ -754,6 +772,23 @@ func (s *instanceService) GetAllInstances(offset, limit int) ([]models.Instance,
 	return instances, total, nil
 }
 
+func hydrateInstancesDesktopStreamProfile(instances []models.Instance) {
+	for idx := range instances {
+		hydrateInstanceDesktopStreamProfile(&instances[idx])
+	}
+}
+
+func hydrateInstanceDesktopStreamProfile(instance *models.Instance) {
+	if instance == nil {
+		return
+	}
+	environmentOverrides, err := parseEnvironmentOverridesJSON(instance.EnvironmentOverridesJSON)
+	if err != nil {
+		return
+	}
+	instance.DesktopStreamProfile = desktopStreamProfileFromEnv(environmentOverrides)
+}
+
 // Start starts an instance
 func (s *instanceService) Start(instanceID int) error {
 	ctx := context.Background()
@@ -1610,6 +1645,22 @@ func (s *instanceService) Update(instanceID int, req UpdateInstanceRequest) erro
 	if req.Description != nil {
 		instance.Description = req.Description
 	}
+	if req.DesktopStreamProfile != nil {
+		profile, ok := normalizeDesktopStreamProfile(*req.DesktopStreamProfile)
+		if !ok || profile == "" {
+			return fmt.Errorf("invalid desktop stream profile")
+		}
+		environmentOverrides, err := parseEnvironmentOverridesJSON(instance.EnvironmentOverridesJSON)
+		if err != nil {
+			return err
+		}
+		environmentOverrides = applyDesktopStreamProfileEnv(environmentOverrides, profile)
+		environmentOverridesJSON, err := marshalEnvironmentOverrides(environmentOverrides)
+		if err != nil {
+			return err
+		}
+		instance.EnvironmentOverridesJSON = environmentOverridesJSON
+	}
 
 	instance.UpdatedAt = time.Now()
 

backend/internal/services/k8s/pvc_service.go

@@ -119,9 +119,12 @@ func (s *PVCService) CreatePVC(ctx context.Context, userID, instanceID int, stor
 		return nil, fmt.Errorf("failed to create PVC %s: %w", pvcName, err)
 	}
 
-	// Wait for PVC to be bound, if not bound within timeout, create PV manually
-	fmt.Printf("PVC %s created, scheduling async binding monitor...\n", pvcName)
-	go s.monitorPVCBinding(context.Background(), namespace, pvcName, userID, instanceID, storageSizeGB, storageClass, 15*time.Second)
+	if usesManualHostPathFallback(storageClass) {
+		fmt.Printf("PVC %s created, scheduling manual hostPath binding monitor...\n", pvcName)
+		go s.monitorPVCBinding(context.Background(), namespace, pvcName, userID, instanceID, storageSizeGB, storageClass, 15*time.Second)
+	} else {
+		fmt.Printf("PVC %s created with dynamic storageClass %s, leaving binding to the provisioner\n", pvcName, storageClass)
+	}
 
 	return createdPVC, nil
 }
@@ -185,7 +188,7 @@ func (s *PVCService) CreateTeamSharedPVC(ctx context.Context, userID, teamID, st
 						return nil, err
 					}
 				}
-				if existingPVC.Status.Phase != corev1.ClaimBound {
+				if existingPVC.Status.Phase != corev1.ClaimBound && usesManualHostPathFallback(storageClass) {
 					go s.monitorTeamSharedPVCBinding(context.Background(), namespace, pvcName, userID, teamID, storageSizeGB, storageClass, 15*time.Second)
 				}
 				return existingPVC, nil
@@ -199,10 +202,16 @@ func (s *PVCService) CreateTeamSharedPVC(ctx context.Context, userID, teamID, st
 			return nil, err
 		}
 	}
-	go s.monitorTeamSharedPVCBinding(context.Background(), namespace, pvcName, userID, teamID, storageSizeGB, storageClass, 15*time.Second)
+	if usesManualHostPathFallback(storageClass) {
+		go s.monitorTeamSharedPVCBinding(context.Background(), namespace, pvcName, userID, teamID, storageSizeGB, storageClass, 15*time.Second)
+	}
 	return createdPVC, nil
 }
 
+func usesManualHostPathFallback(storageClass string) bool {
+	return strings.EqualFold(strings.TrimSpace(storageClass), "manual")
+}
+
 func (s *PVCService) monitorTeamSharedPVCBinding(ctx context.Context, namespace, pvcName string, userID, teamID, storageSizeGB int, storageClass string, timeout time.Duration) {
 	if _, err := s.waitForTeamSharedPVCBinding(ctx, namespace, pvcName, userID, teamID, storageSizeGB, storageClass, timeout); err != nil {
 		fmt.Printf("Async Team shared PVC binding monitor failed for %s: %v\n", pvcName, err)
@@ -225,6 +234,10 @@ func (s *PVCService) waitForTeamSharedPVCBinding(ctx context.Context, namespace,
 	for {
 		select {
 		case <-timeoutChan:
+			if !usesManualHostPathFallback(storageClass) {
+				fmt.Printf("Team shared PVC %s binding timeout with dynamic storageClass %s, leaving binding to the provisioner\n", pvcName, storageClass)
+				return pvc, nil
+			}
 			fmt.Printf("Team shared PVC %s binding timeout, creating hostPath RWX PV manually\n", pvcName)
 			return s.createPVForTeamSharedPVC(ctx, namespace, pvcName, userID, teamID, storageSizeGB, storageClass)
 		case <-ticker.C:
@@ -629,7 +642,11 @@ func (s *PVCService) waitForPVCBinding(ctx context.Context, namespace, pvcName s
 	for {
 		select {
 		case <-timeoutChan:
-			// Timeout, try to create PV manually
+			if !usesManualHostPathFallback(storageClass) {
+				fmt.Printf("PVC %s binding timeout with dynamic storageClass %s, leaving binding to the provisioner\n", pvcName, storageClass)
+				return pvc, nil
+			}
+			// Timeout, try to create PV manually.
 			fmt.Printf("PVC %s binding timeout, creating PV manually\n", pvcName)
 			return s.createPVForPVC(ctx, namespace, pvcName, userID, instanceID, storageSizeGB, storageClass)
 		case <-ticker.C:

backend/internal/services/k8s/pvc_service_test.go

@@ -6,6 +6,7 @@ import (
 	"path/filepath"
 	"strings"
 	"testing"
+	"time"
 
 	corev1 "k8s.io/api/core/v1"
 	storagev1 "k8s.io/api/storage/v1"
@@ -456,6 +457,45 @@ func TestNodeSelectorForPVCFallsBackForNoProvisionerStorageClass(t *testing.T) {
 	}
 }
 
+func TestWaitForPVCBindingLeavesDynamicStorageClassToProvisioner(t *testing.T) {
+	pvc := &corev1.PersistentVolumeClaim{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "clawreef-114-pvc",
+			Namespace: "clawmanager-user-155",
+		},
+		Spec: corev1.PersistentVolumeClaimSpec{
+			StorageClassName: stringPtr("local-path"),
+		},
+		Status: corev1.PersistentVolumeClaimStatus{
+			Phase: corev1.ClaimPending,
+		},
+	}
+	service := &PVCService{
+		client: &Client{
+			Clientset: fake.NewSimpleClientset(pvc),
+		},
+	}
+
+	got, err := service.waitForPVCBinding(context.Background(), pvc.Namespace, pvc.Name, 155, 114, 100, "local-path", time.Nanosecond)
+	if err != nil {
+		t.Fatalf("waitForPVCBinding returned error: %v", err)
+	}
+	if got.Name != pvc.Name {
+		t.Fatalf("expected PVC %q, got %q", pvc.Name, got.Name)
+	}
+	pvs, err := service.client.Clientset.CoreV1().PersistentVolumes().List(context.Background(), metav1.ListOptions{})
+	if err != nil {
+		t.Fatalf("failed to list PVs: %v", err)
+	}
+	if len(pvs.Items) != 0 {
+		t.Fatalf("expected no manual PVs for dynamic storageClass, got %#v", pvs.Items)
+	}
+}
+
+func stringPtr(value string) *string {
+	return &value
+}
+
 func requireHostnameAffinity(t *testing.T, affinity *corev1.VolumeNodeAffinity, hostname string) {
 	t.Helper()
 	if affinity == nil || affinity.Required == nil || len(affinity.Required.NodeSelectorTerms) != 1 {

backend/internal/services/leader/leader.go

@@ -0,0 +1,124 @@
+// Package leader wraps Kubernetes lease-based leader election so that
+// control-plane singleton background loops (the K8s sync loop, Team event
+// consumers and the stale-task monitor) run on exactly one clawmanager-app
+// replica at a time. The HTTP API and the in-pod nginx desktop data plane run
+// on every replica; only these background loops must be gated.
+package leader
+
+import (
+	"context"
+	"log"
+	"os"
+	"time"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/tools/leaderelection"
+	"k8s.io/client-go/tools/leaderelection/resourcelock"
+)
+
+// Config holds the resolved leader-election parameters.
+type Config struct {
+	Namespace     string
+	LeaseName     string
+	Identity      string
+	LeaseDuration time.Duration
+	RenewDeadline time.Duration
+	RetryPeriod   time.Duration
+}
+
+// Callbacks are invoked when this replica acquires or loses leadership.
+//
+// OnStartedLeading receives a context that is cancelled when leadership is
+// lost; callers may use it to scope leader-only work. OnStoppedLeading is
+// invoked when leadership is lost (or the parent context is cancelled) and must
+// stop any work started in OnStartedLeading. Both callbacks must be safe to
+// call repeatedly across leadership transitions.
+type Callbacks struct {
+	OnStartedLeading func(ctx context.Context)
+	OnStoppedLeading func()
+}
+
+// Run participates in leader election and blocks until ctx is cancelled. It is
+// resilient to leadership loss: after losing the lease it re-participates so a
+// replica that briefly lost leadership can re-acquire it and restart the
+// background loops. Run is intended to be called in its own goroutine.
+func Run(ctx context.Context, clientset kubernetes.Interface, cfg Config, cb Callbacks) {
+	identity := cfg.Identity
+	if identity == "" {
+		if host, err := os.Hostname(); err == nil && host != "" {
+			identity = host
+		} else {
+			identity = "clawmanager-app"
+		}
+	}
+
+	leaseDuration := cfg.LeaseDuration
+	if leaseDuration <= 0 {
+		leaseDuration = 15 * time.Second
+	}
+	renewDeadline := cfg.RenewDeadline
+	if renewDeadline <= 0 || renewDeadline >= leaseDuration {
+		renewDeadline = leaseDuration * 2 / 3
+	}
+	retryPeriod := cfg.RetryPeriod
+	if retryPeriod <= 0 {
+		retryPeriod = 2 * time.Second
+	}
+
+	lock := &resourcelock.LeaseLock{
+		LeaseMeta: metav1.ObjectMeta{
+			Name:      cfg.LeaseName,
+			Namespace: cfg.Namespace,
+		},
+		Client: clientset.CoordinationV1(),
+		LockConfig: resourcelock.ResourceLockConfig{
+			Identity: identity,
+		},
+	}
+
+	electionCfg := leaderelection.LeaderElectionConfig{
+		Lock:            lock,
+		ReleaseOnCancel: true,
+		LeaseDuration:   leaseDuration,
+		RenewDeadline:   renewDeadline,
+		RetryPeriod:     retryPeriod,
+		Callbacks: leaderelection.LeaderCallbacks{
+			OnStartedLeading: func(c context.Context) {
+				log.Printf("[leader] acquired leadership (identity=%s lease=%s/%s)", identity, cfg.Namespace, cfg.LeaseName)
+				if cb.OnStartedLeading != nil {
+					cb.OnStartedLeading(c)
+				}
+			},
+			OnStoppedLeading: func() {
+				log.Printf("[leader] lost leadership (identity=%s)", identity)
+				if cb.OnStoppedLeading != nil {
+					cb.OnStoppedLeading()
+				}
+			},
+		},
+	}
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		default:
+		}
+
+		elector, err := leaderelection.NewLeaderElector(electionCfg)
+		if err != nil {
+			log.Printf("[leader] failed to create leader elector: %v; retrying in %s", err, retryPeriod)
+			select {
+			case <-ctx.Done():
+				return
+			case <-time.After(retryPeriod):
+				continue
+			}
+		}
+
+		// Blocks until leadership is lost or ctx is cancelled. On loss we loop
+		// and re-participate.
+		elector.Run(ctx)
+	}
+}

backend/internal/services/sync_service.go

@@ -3,6 +3,7 @@ package services
 import (
 	"context"
 	"fmt"
+	"sync"
 	"time"
 
 	"clawreef/internal/models"
@@ -20,7 +21,10 @@ type SyncService struct {
 	podService           *k8s.PodService
 	deploymentService    *k8s.InstanceDeploymentService
 	interval             time.Duration
-	stopChan             chan struct{}
+
+	mu       sync.Mutex
+	running  bool
+	stopChan chan struct{}
 }
 
 // NewSyncService creates a new sync service
@@ -31,23 +35,40 @@ func NewSyncService(instanceRepo repository.InstanceRepository, runtimeStatusSer
 		podService:           k8s.NewPodService(),
 		deploymentService:    k8s.NewInstanceDeploymentService(),
 		interval:             5 * time.Second, // Sync every 5 seconds for more responsive status updates
-		stopChan:             make(chan struct{}),
 	}
 }
 
-// Start starts the sync service
+// Start starts the sync loop. It is safe to call repeatedly: a second call
+// while already running is a no-op, and after Stop the service can be started
+// again (used by leader-election re-acquisition).
 func (s *SyncService) Start() {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if s.running {
+		return
+	}
+	s.stopChan = make(chan struct{})
+	s.running = true
 	fmt.Println("Starting K8s state sync service...")
-	go s.syncLoop()
+	go s.syncLoop(s.stopChan)
 }
 
-// Stop stops the sync service
+// Stop stops the sync loop. It is idempotent: calling Stop when not running is
+// a no-op, and it never closes the same channel twice.
 func (s *SyncService) Stop() {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if !s.running {
+		return
+	}
 	close(s.stopChan)
+	s.running = false
 }
 
-// syncLoop runs the synchronization loop
-func (s *SyncService) syncLoop() {
+// syncLoop runs the synchronization loop until its stop channel is closed. The
+// channel is passed in (rather than read from the struct) so a restarted loop
+// never races with a previously stopped one.
+func (s *SyncService) syncLoop(stop <-chan struct{}) {
 	fmt.Printf("[SyncService] Starting sync loop with interval %v\n", s.interval)
 	ticker := time.NewTicker(s.interval)
 	defer ticker.Stop()
@@ -62,7 +83,7 @@ func (s *SyncService) syncLoop() {
 		case <-ticker.C:
 			fmt.Println("[SyncService] Tick - running scheduled sync...")
 			s.syncAllInstances()
-		case <-s.stopChan:
+		case <-stop:
 			fmt.Println("[SyncService] Stopping K8s state sync service...")
 			return
 		}

backend/internal/services/team_service.go

@@ -29,6 +29,7 @@ const (
 
 	defaultTeamTaskStaleTimeout = 30 * time.Minute
 	teamTaskStaleSweepInterval  = 30 * time.Second
+	teamConsumerScanInterval    = 10 * time.Second
 
 	initialLeaderTaskIntent = "team_bootstrap_introduction"
 	teamTaskCompletionTool  = "team_complete_task"
@@ -42,8 +43,8 @@ var (
 )
 
 type TeamService interface {
-	Start()
-	Stop()
+	StartBackground(ctx context.Context)
+	StopBackground()
 	CreateTeam(userID int, req CreateTeamRequest) (*TeamDetailsPayload, error)
 	ListTeams(userID, offset, limit int) (*TeamListPayload, error)
 	GetTeam(userID, teamID int) (*TeamDetailsPayload, error)
@@ -137,6 +138,8 @@ type teamService struct {
 	ctx                  context.Context
 	cancel               context.CancelFunc
 	mu                   sync.Mutex
+	running              bool
+	wg                   sync.WaitGroup
 	consumers            map[int]struct{}
 	staleMonitorStarted  bool
 	runtimeWorkspaceRoot string
@@ -188,20 +191,93 @@ func NewTeamService(repo repository.TeamRepository, instanceService InstanceServ
 	return service
 }
 
-func (s *teamService) Start() {
-	teams, err := s.repo.ListActiveTeams()
-	if err != nil {
-		fmt.Printf("Warning: failed to start Team event consumers: %v\n", err)
+// StartBackground starts the leader-only background workers: a periodic scan
+// that ensures a Redis event consumer is running for every active team, and
+// the stale-task monitor. It is safe to call repeatedly (a second call while
+// running is a no-op) and can be called again after StopBackground, which is
+// required for leader-election re-acquisition. HTTP request handling does not
+// depend on these workers, so followers can still serve the API and the in-pod
+// nginx data plane while only the leader runs them.
+func (s *teamService) StartBackground(parent context.Context) {
+	s.mu.Lock()
+	if s.running {
+		s.mu.Unlock()
 		return
 	}
-	for _, team := range teams {
-		s.ensureConsumer(team.ID)
+	if parent == nil {
+		parent = context.Background()
 	}
-	s.ensureStaleTaskMonitor()
+	ctx, cancel := context.WithCancel(parent)
+	s.ctx = ctx
+	s.cancel = cancel
+	s.running = true
+	s.staleMonitorStarted = false
+	s.consumers = map[int]struct{}{}
+	s.wg.Add(1)
+	go s.consumerScanLoop(ctx)
+	s.mu.Unlock()
+
+	fmt.Println("[TeamService] Starting leader-only background workers...")
+	s.ensureStaleTaskMonitor(ctx)
 }
 
-func (s *teamService) Stop() {
-	s.cancel()
+// StopBackground stops all background workers and blocks until they have fully
+// exited, so a subsequent StartBackground starts from a clean state with no
+// goroutines from the previous generation still touching shared maps. It is
+// idempotent.
+func (s *teamService) StopBackground() {
+	s.mu.Lock()
+	if !s.running {
+		s.mu.Unlock()
+		return
+	}
+	s.running = false
+	cancel := s.cancel
+	s.mu.Unlock()
+
+	fmt.Println("[TeamService] Stopping leader-only background workers...")
+	if cancel != nil {
+		cancel()
+	}
+	s.wg.Wait()
+
+	s.mu.Lock()
+	s.consumers = map[int]struct{}{}
+	s.staleMonitorStarted = false
+	s.mu.Unlock()
+}
+
+// consumerScanLoop periodically ensures a consumer goroutine exists for every
+// active team. Team creation no longer starts consumers inline (that would run
+// on whichever replica served the request); the leader picks up newly active
+// teams here within teamConsumerScanInterval.
+func (s *teamService) consumerScanLoop(ctx context.Context) {
+	defer s.wg.Done()
+
+	s.ensureConsumersForActiveTeams(ctx)
+
+	ticker := time.NewTicker(teamConsumerScanInterval)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			s.ensureConsumersForActiveTeams(ctx)
+		}
+	}
+}
+
+func (s *teamService) ensureConsumersForActiveTeams(ctx context.Context) {
+	teams, err := s.repo.ListActiveTeams()
+	if err != nil {
+		fmt.Printf("Warning: failed to list active teams for consumer scan: %v\n", err)
+		return
+	}
+	for i := range teams {
+		s.ensureConsumer(ctx, teams[i].ID)
+	}
 }
 
 func (s *teamService) CreateTeam(userID int, req CreateTeamRequest) (*TeamDetailsPayload, error) {
@@ -306,8 +382,10 @@ func (s *teamService) CreateTeam(userID int, req CreateTeamRequest) (*TeamDetail
 	if err := s.repo.UpdateTeam(team); err != nil {
 		return nil, err
 	}
-	s.ensureConsumer(team.ID)
-	s.ensureStaleTaskMonitor()
+	// Background consumers / stale-task monitor are leader-only and started by
+	// the leader's periodic scan (consumerScanLoop). Starting them here would
+	// run them on whichever replica served the create request, bypassing
+	// leader election, so we intentionally do not call ensureConsumer here.
 	if err := s.dispatchInitialLeaderTask(userID, team); err != nil {
 		fmt.Printf("Warning: failed to dispatch initial Team %d leader task: %v\n", team.ID, err)
 		if recordErr := s.recordInitialLeaderTaskDispatchFailure(team.ID, err); recordErr != nil {
@@ -962,17 +1040,22 @@ func (s *teamService) requireOwnedTeam(userID, teamID int) (*models.Team, error)
 	return team, nil
 }
 
-func (s *teamService) ensureConsumer(teamID int) {
+func (s *teamService) ensureConsumer(ctx context.Context, teamID int) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
+	if !s.running {
+		return
+	}
 	if _, exists := s.consumers[teamID]; exists {
 		return
 	}
 	s.consumers[teamID] = struct{}{}
-	go s.consumeTeamEvents(teamID)
+	s.wg.Add(1)
+	go s.consumeTeamEvents(ctx, teamID)
 }
 
-func (s *teamService) consumeTeamEvents(teamID int) {
+func (s *teamService) consumeTeamEvents(ctx context.Context, teamID int) {
+	defer s.wg.Done()
 	defer func() {
 		s.mu.Lock()
 		delete(s.consumers, teamID)
@@ -981,7 +1064,7 @@ func (s *teamService) consumeTeamEvents(teamID int) {
 
 	for {
 		select {
-		case <-s.ctx.Done():
+		case <-ctx.Done():
 			return
 		default:
 		}
@@ -991,7 +1074,7 @@ func (s *teamService) consumeTeamEvents(teamID int) {
 			time.Sleep(5 * time.Second)
 			continue
 		}
-		bus, err := s.redisBusForTeam(s.ctx, team)
+		bus, err := s.redisBusForTeam(ctx, team)
 		if err != nil {
 			time.Sleep(5 * time.Second)
 			continue
@@ -1000,7 +1083,7 @@ func (s *teamService) consumeTeamEvents(teamID int) {
 		if lastID == "" {
 			lastID = "0-0"
 		}
-		messages, err := bus.XRead(s.ctx, teamEventsKey(teamID), lastID, 5*time.Second)
+		messages, err := bus.XRead(ctx, teamEventsKey(teamID), lastID, 5*time.Second)
 		if err != nil {
 			time.Sleep(2 * time.Second)
 			continue
@@ -1016,23 +1099,28 @@ func (s *teamService) consumeTeamEvents(teamID int) {
 	}
 }
 
-func (s *teamService) ensureStaleTaskMonitor() {
+func (s *teamService) ensureStaleTaskMonitor(ctx context.Context) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
+	if !s.running {
+		return
+	}
 	if s.staleMonitorStarted {
 		return
 	}
 	s.staleMonitorStarted = true
-	go s.monitorStaleTasks()
+	s.wg.Add(1)
+	go s.monitorStaleTasks(ctx)
 }
 
-func (s *teamService) monitorStaleTasks() {
+func (s *teamService) monitorStaleTasks(ctx context.Context) {
+	defer s.wg.Done()
 	ticker := time.NewTicker(teamTaskStaleSweepInterval)
 	defer ticker.Stop()
 
 	for {
 		select {
-		case <-s.ctx.Done():
+		case <-ctx.Done():
 			return
 		case <-ticker.C:
 			if err := s.sweepStaleTasks(); err != nil {

deployments/container/start.sh

@@ -96,6 +96,15 @@ esac
 
 sed -i "s/client_max_body_size [0-9][0-9]*m;/client_max_body_size ${CLAWMANAGER_WORKSPACE_ARCHIVE_MAX_MIB}m;/" /etc/nginx/nginx.conf
 
+# Resolve the cluster DNS server for nginx so the desktop location can resolve
+# per-instance Service FQDNs at request time. Prefer the first nameserver from
+# /etc/resolv.conf, falling back to the common in-cluster DNS ClusterIP.
+DNS_RESOLVER="$(awk '/^nameserver/ {print $2; exit}' /etc/resolv.conf 2>/dev/null || true)"
+if [ -z "${DNS_RESOLVER}" ]; then
+  DNS_RESOLVER="10.96.0.10"
+fi
+sed -i "s/__DNS_RESOLVER__/${DNS_RESOLVER}/g" /etc/nginx/nginx.conf
+
 /usr/local/bin/clawreef-server &
 backend_pid=$!
 

deployments/k3s/clawmanager.yaml

@@ -979,6 +979,33 @@ subjects:
     name: clawmanager-app
     namespace: clawmanager-system
 ---
+# Explicit lease permissions for control-plane leader election. The
+# cluster-admin binding above already covers this; this Role documents the
+# requirement and keeps leader election working if RBAC is tightened later.
+apiVersion: rbac.authorization.k8s.io/v1
+kind: Role
+metadata:
+  name: clawmanager-app-leaderelection
+  namespace: clawmanager-system
+rules:
+  - apiGroups: ["coordination.k8s.io"]
+    resources: ["leases"]
+    verbs: ["get", "list", "watch", "create", "update"]
+---
+apiVersion: rbac.authorization.k8s.io/v1
+kind: RoleBinding
+metadata:
+  name: clawmanager-app-leaderelection
+  namespace: clawmanager-system
+roleRef:
+  apiGroup: rbac.authorization.k8s.io
+  kind: Role
+  name: clawmanager-app-leaderelection
+subjects:
+  - kind: ServiceAccount
+    name: clawmanager-app
+    namespace: clawmanager-system
+---
 apiVersion: apps/v1
 kind: Deployment
 metadata:
@@ -1039,6 +1066,21 @@ spec:
               value: "redis://clawmanager-redis.clawmanager-system.svc.cluster.local:6379/0"
             - name: TEAM_REDIS_URL
               value: "redis://clawmanager-redis.clawmanager-system.svc.cluster.local:6379/0"
+            - name: CLAWMANAGER_DESKTOP_DIRECT_PROXY
+              value: "true"
+            # Leader election for control-plane singleton background loops.
+            # Required before scaling replicas > 1 so SyncService / Team event
+            # consumers / stale-task monitor run on exactly one replica.
+            - name: CLAWMANAGER_LEADER_ELECTION
+              value: "true"
+            - name: POD_NAME
+              valueFrom:
+                fieldRef:
+                  fieldPath: metadata.name
+            - name: POD_NAMESPACE
+              valueFrom:
+                fieldRef:
+                  fieldPath: metadata.namespace
             - name: CLAWMANAGER_TEAM_REDIS_URL
               value: "redis://clawmanager-redis.clawmanager-system.svc.cluster.local:6379/0"
             - name: CLAWMANAGER_TEAM_MANAGER_BASE_URL

deployments/k8s/clawmanager.yaml

@@ -1116,6 +1116,33 @@ subjects:
     name: clawmanager-app
     namespace: clawmanager-system
 ---
+# Explicit lease permissions for control-plane leader election. The
+# cluster-admin binding above already covers this; this Role documents the
+# requirement and keeps leader election working if RBAC is tightened later.
+apiVersion: rbac.authorization.k8s.io/v1
+kind: Role
+metadata:
+  name: clawmanager-app-leaderelection
+  namespace: clawmanager-system
+rules:
+  - apiGroups: ["coordination.k8s.io"]
+    resources: ["leases"]
+    verbs: ["get", "list", "watch", "create", "update"]
+---
+apiVersion: rbac.authorization.k8s.io/v1
+kind: RoleBinding
+metadata:
+  name: clawmanager-app-leaderelection
+  namespace: clawmanager-system
+roleRef:
+  apiGroup: rbac.authorization.k8s.io
+  kind: Role
+  name: clawmanager-app-leaderelection
+subjects:
+  - kind: ServiceAccount
+    name: clawmanager-app
+    namespace: clawmanager-system
+---
 apiVersion: apps/v1
 kind: Deployment
 metadata:
@@ -1255,6 +1282,21 @@ spec:
               value: "redis://clawmanager-redis.clawmanager-system.svc.cluster.local:6379/0"
             - name: TEAM_REDIS_URL
               value: "redis://clawmanager-redis.clawmanager-system.svc.cluster.local:6379/0"
+            - name: CLAWMANAGER_DESKTOP_DIRECT_PROXY
+              value: "true"
+            # Leader election for control-plane singleton background loops.
+            # Required before scaling replicas > 1 so SyncService / Team event
+            # consumers / stale-task monitor run on exactly one replica.
+            - name: CLAWMANAGER_LEADER_ELECTION
+              value: "true"
+            - name: POD_NAME
+              valueFrom:
+                fieldRef:
+                  fieldPath: metadata.name
+            - name: POD_NAMESPACE
+              valueFrom:
+                fieldRef:
+                  fieldPath: metadata.namespace
             - name: CLAWMANAGER_TEAM_REDIS_URL
               value: "redis://clawmanager-redis.clawmanager-system.svc.cluster.local:6379/0"
             - name: CLAWMANAGER_TEAM_MANAGER_BASE_URL

deployments/nginx/nginx.conf

@@ -1,5 +1,13 @@
+load_module modules/ngx_http_js_module.so;
+
 worker_processes auto;
 
+# Expose the instance-access JWT secret to njs (process.env) so the edge
+# gateway can verify desktop access tokens locally. Values come from the
+# container environment; they are not logged.
+env INSTANCE_ACCESS_TOKEN_SECRET;
+env JWT_SECRET;
+
 events {
     worker_connections 8192;
 }
@@ -8,6 +16,12 @@ http {
     include /etc/nginx/mime.types;
     default_type application/octet-stream;
 
+    js_import desktop from /etc/nginx/njs/desktop_auth.js;
+
+    # Cluster DNS resolver, templated by start.sh from /etc/resolv.conf so the
+    # desktop location can resolve per-instance Service FQDNs at request time.
+    resolver __DNS_RESOLVER__ valid=10s ipv6=off;
+
     sendfile on;
     tcp_nopush on;
     tcp_nodelay on;
@@ -34,6 +48,16 @@ http {
         '' close;
     }
 
+    map $desktop_target $desktop_proxy_mode {
+        "deny" "deny";
+        "http://127.0.0.1:9001" "fallback";
+        default "direct";
+    }
+
+    log_format desktop_proxy '$remote_addr - $status "$request_method $desktop_clean_uri $server_protocol" '
+                             'mode=$desktop_proxy_mode upstream="$upstream_addr" '
+                             'rt=$request_time urt=$upstream_response_time bytes=$body_bytes_sent';
+
     upstream clawreef_backend {
         server 127.0.0.1:9001;
         keepalive 128;
@@ -104,20 +128,45 @@ http {
             proxy_request_buffering off;
         }
 
-        location ~ ^/api/v1/instances/[0-9]+/proxy/? {
-            proxy_pass http://clawreef_backend;
+        location ~ ^/api/v1/instances/(?<inst_id>[0-9]+)/proxy(?<proxy_sub>/.*)?$ {
+            # njs verifies the instance-access JWT locally and chooses the
+            # proxy target: the instance Service (direct), the in-process
+            # control-plane proxy (gray fallback), or "deny".
+            js_set $desktop_target desktop.resolveTarget;
+            # Forward the original URI to the upstream but with the JWT "token"
+            # query param stripped, so the access token never reaches webtop.
+            js_set $desktop_clean_uri desktop.cleanUri;
+
+            access_log /var/log/nginx/access.log desktop_proxy;
+            add_header X-ClawManager-Desktop-Proxy $desktop_proxy_mode always;
+
+            error_page 463 = @desktop_denied;
+            if ($desktop_target = "deny") {
+                return 463;
+            }
+
             proxy_http_version 1.1;
-            proxy_set_header Host $host;
+            proxy_set_header Host $http_host;
             proxy_set_header X-Real-IP $remote_addr;
             proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
             proxy_set_header X-Forwarded-Proto https;
+            proxy_set_header X-Forwarded-Prefix /api/v1/instances/$inst_id/proxy;
             proxy_set_header Upgrade $http_upgrade;
             proxy_set_header Connection $connection_upgrade;
+            # Browsers may include the original iframe URL (with ?token=...)
+            # as Referer on subresource requests. Do not forward it to webtop.
+            proxy_set_header Referer "";
+
+            proxy_ssl_verify off;
+            proxy_ssl_server_name on;
+
             proxy_connect_timeout 15s;
             proxy_read_timeout 3600s;
             proxy_send_timeout 3600s;
             proxy_buffering off;
             proxy_request_buffering off;
+
+            proxy_pass $desktop_target$desktop_clean_uri;
         }
 
         location /api/ {
@@ -136,5 +185,10 @@ http {
         location / {
             try_files $uri $uri/ /index.html;
         }
+
+        location @desktop_denied {
+            add_header Content-Type text/plain always;
+            return 401 "Access token expired or invalid";
+        }
     }
 }

deployments/nginx/njs/desktop_auth.js

@@ -0,0 +1,145 @@
+// desktop_auth.js
+//
+// njs helper used by the in-pod nginx to decide where a desktop proxy request
+// should be sent. It locally verifies the ClawManager instance-access JWT
+// (HS256, same secret as the Go control plane) and returns a proxy_pass target:
+//
+//   "deny"                     -> token missing/invalid/expired/mismatched
+//   "https://<host:port>"      -> direct connection to the instance Service
+//                                 (taken from the token's "upstream" claim)
+//   "http://127.0.0.1:9001"    -> fall back to the in-process control-plane
+//                                 proxy (gray rollout / legacy tokens without
+//                                 an "upstream" claim)
+//
+// The function runs in the main request context (js_set), so query args and
+// cookies of the original request are available.
+
+var crypto = require('crypto');
+
+var CONTROL_PLANE_FALLBACK = 'http://127.0.0.1:9001';
+var DENY = 'deny';
+
+function secret() {
+    return process.env.INSTANCE_ACCESS_TOKEN_SECRET || process.env.JWT_SECRET || '';
+}
+
+// Normalize base64url / base64 to pad-less standard base64 for comparison.
+function toStdB64NoPad(s) {
+    return String(s).replace(/-/g, '+').replace(/_/g, '/').replace(/=+$/, '');
+}
+
+function b64urlToString(s) {
+    var std = String(s).replace(/-/g, '+').replace(/_/g, '/');
+    while (std.length % 4 !== 0) {
+        std += '=';
+    }
+    return Buffer.from(std, 'base64').toString();
+}
+
+function readToken(r) {
+    if (r.args && r.args.token) {
+        return r.args.token;
+    }
+
+    var cookie = r.headersIn['Cookie'];
+    if (!cookie) {
+        return '';
+    }
+
+    var name = 'instance_access_' + r.variables.inst_id;
+    var parts = cookie.split(';');
+    for (var i = 0; i < parts.length; i++) {
+        var kv = parts[i].trim();
+        var eq = kv.indexOf('=');
+        if (eq > 0 && kv.substring(0, eq) === name) {
+            return kv.substring(eq + 1);
+        }
+    }
+    return '';
+}
+
+function resolveTarget(r) {
+    var key = secret();
+    if (!key) {
+        r.error('desktop_auth: missing JWT secret in environment');
+        return DENY;
+    }
+
+    var token = readToken(r);
+    if (!token) {
+        return DENY;
+    }
+
+    var segments = token.split('.');
+    if (segments.length !== 3) {
+        return DENY;
+    }
+
+    var signingInput = segments[0] + '.' + segments[1];
+    var expected = crypto.createHmac('sha256', key).update(signingInput).digest('base64');
+    if (toStdB64NoPad(expected) !== toStdB64NoPad(segments[2])) {
+        return DENY;
+    }
+
+    var payload;
+    try {
+        payload = JSON.parse(b64urlToString(segments[1]));
+    } catch (e) {
+        return DENY;
+    }
+
+    if (payload.token_type !== 'instance_access') {
+        return DENY;
+    }
+
+    if (payload.exp && (Date.now() / 1000) >= Number(payload.exp)) {
+        return DENY;
+    }
+
+    if (String(payload.instance_id) !== String(r.variables.inst_id)) {
+        return DENY;
+    }
+
+    if (payload.upstream) {
+        return 'https://' + payload.upstream;
+    }
+
+    return CONTROL_PLANE_FALLBACK;
+}
+
+// cleanUri returns the original request URI (path + query) with the "token"
+// query parameter stripped, so the instance-access JWT is never forwarded to
+// the upstream desktop (it would otherwise land in webtop/KasmVNC access logs).
+// njs validation in resolveTarget still reads the original token from the
+// request, so authentication is unaffected. Subresource / WebSocket requests
+// carry no token and are returned unchanged.
+function cleanUri(r) {
+    var uri = r.variables.request_uri || r.uri || '/';
+    var q = uri.indexOf('?');
+    if (q < 0) {
+        return uri;
+    }
+
+    var path = uri.substring(0, q);
+    var query = uri.substring(q + 1);
+    var parts = query.split('&');
+    var kept = [];
+    for (var i = 0; i < parts.length; i++) {
+        if (parts[i] === '') {
+            continue;
+        }
+        var eq = parts[i].indexOf('=');
+        var name = eq < 0 ? parts[i] : parts[i].substring(0, eq);
+        if (name === 'token') {
+            continue;
+        }
+        kept.push(parts[i]);
+    }
+
+    if (kept.length === 0) {
+        return path;
+    }
+    return path + '?' + kept.join('&');
+}
+
+export default { resolveTarget, cleanUri };

frontend/src/hooks/useInstanceDesktopAccess.ts

@@ -256,7 +256,7 @@ export function useInstanceDesktopAccess({
           return;
         }
 
-        const nextEmbedUrl = resolveEmbedUrl(data.proxy_url || data.access_url);
+        const nextEmbedUrl = resolveEmbedUrl(data.access_url || data.proxy_url);
         const nextExpiresAt = new Date(data.expires_at);
         const previousEmbedUrl = embedUrlRef.current;
 

frontend/src/lib/i18n.ts

@@ -1860,6 +1860,24 @@ export const translations: Record<Locale, TranslationTree> = {
       presetCpuShort: "CPU",
       presetRamShort: "RAM",
       presetDiskShort: "Disk",
+      desktopStreamProfile: "Desktop Stream Profile",
+      desktopStreamSettings: "Stream Settings",
+      desktopStreamProfileDesc:
+        "Choose the Selkies desktop streaming balance. Resolution remains adaptive to the browser window.",
+      desktopStreamLow: "Low",
+      desktopStreamLowDesc:
+        "Best for weak networks and large concurrency. Keeps motion usable while reducing clarity.",
+      desktopStreamStandard: "Standard",
+      desktopStreamStandardDesc:
+        "Balanced default with responsive motion and moderate clarity.",
+      desktopStreamHigh: "High",
+      desktopStreamHighDesc:
+        "Smoother motion with higher bandwidth and CPU usage.",
+      restartRequiredAfterChange:
+        "Changes take effect after the instance is restarted.",
+      desktopStreamSavedRestartRequired:
+        "Stream profile saved. Restart the instance for it to take effect.",
+      desktopStreamSaveFailed: "Failed to save stream profile.",
       environmentVariables: "Environment Variables",
       clawManagerBuiltIns: "ClawManager Built-ins",
       hideBuiltIns: "Hide Built-ins",
@@ -3126,6 +3144,19 @@ export const translations: Record<Locale, TranslationTree> = {
       presetCpuShort: "CPU",
       presetRamShort: "RAM",
       presetDiskShort: "磁盘",
+      desktopStreamProfile: "桌面流档位",
+      desktopStreamSettings: "桌面流设置",
+      desktopStreamProfileDesc:
+        "选择 Selkies 桌面流的流畅度与带宽平衡。分辨率仍会跟随浏览器窗口动态适配。",
+      desktopStreamLow: "低带宽",
+      desktopStreamLowDesc: "适合弱网和高并发场景。保留可用帧率，降低清晰度。",
+      desktopStreamStandard: "标准",
+      desktopStreamStandardDesc: "适合日常桌面操作，兼顾跟手感和清晰度。",
+      desktopStreamHigh: "高清",
+      desktopStreamHighDesc: "画面更顺滑，但会占用更多带宽和 CPU。",
+      restartRequiredAfterChange: "修改会在实例重启后生效。",
+      desktopStreamSavedRestartRequired: "桌面流档位已保存，请重启实例后生效。",
+      desktopStreamSaveFailed: "保存桌面流档位失败。",
       environmentVariables: "环境变量",
       clawManagerBuiltIns: "ClawManager 内置变量",
       hideBuiltIns: "收起内置变量",
@@ -4373,6 +4404,24 @@ export const translations: Record<Locale, TranslationTree> = {
       presetCpuShort: "CPU",
       presetRamShort: "RAM",
       presetDiskShort: "ディスク",
+      desktopStreamProfile: "Desktop Stream Profile",
+      desktopStreamSettings: "Stream Settings",
+      desktopStreamProfileDesc:
+        "Choose the Selkies desktop streaming balance. Resolution remains adaptive to the browser window.",
+      desktopStreamLow: "Low",
+      desktopStreamLowDesc:
+        "Best for weak networks and large concurrency. Keeps motion usable while reducing clarity.",
+      desktopStreamStandard: "Standard",
+      desktopStreamStandardDesc:
+        "Balanced default with responsive motion and moderate clarity.",
+      desktopStreamHigh: "High",
+      desktopStreamHighDesc:
+        "Smoother motion with higher bandwidth and CPU usage.",
+      restartRequiredAfterChange:
+        "Changes take effect after the instance is restarted.",
+      desktopStreamSavedRestartRequired:
+        "Stream profile saved. Restart the instance for it to take effect.",
+      desktopStreamSaveFailed: "Failed to save stream profile.",
       environmentVariables: "環境変数",
       clawManagerBuiltIns: "ClawManager 組み込み変数",
       hideBuiltIns: "組み込み変数を隠す",
@@ -5630,6 +5679,24 @@ export const translations: Record<Locale, TranslationTree> = {
       presetCpuShort: "CPU",
       presetRamShort: "RAM",
       presetDiskShort: "디스크",
+      desktopStreamProfile: "Desktop Stream Profile",
+      desktopStreamSettings: "Stream Settings",
+      desktopStreamProfileDesc:
+        "Choose the Selkies desktop streaming balance. Resolution remains adaptive to the browser window.",
+      desktopStreamLow: "Low",
+      desktopStreamLowDesc:
+        "Best for weak networks and large concurrency. Keeps motion usable while reducing clarity.",
+      desktopStreamStandard: "Standard",
+      desktopStreamStandardDesc:
+        "Balanced default with responsive motion and moderate clarity.",
+      desktopStreamHigh: "High",
+      desktopStreamHighDesc:
+        "Smoother motion with higher bandwidth and CPU usage.",
+      restartRequiredAfterChange:
+        "Changes take effect after the instance is restarted.",
+      desktopStreamSavedRestartRequired:
+        "Stream profile saved. Restart the instance for it to take effect.",
+      desktopStreamSaveFailed: "Failed to save stream profile.",
       environmentVariables: "환경 변수",
       clawManagerBuiltIns: "ClawManager 내장 변수",
       hideBuiltIns: "내장 변수 숨기기",
@@ -6912,6 +6979,24 @@ export const translations: Record<Locale, TranslationTree> = {
       presetCpuShort: "CPU",
       presetRamShort: "RAM",
       presetDiskShort: "Disk",
+      desktopStreamProfile: "Desktop Stream Profile",
+      desktopStreamSettings: "Stream Settings",
+      desktopStreamProfileDesc:
+        "Choose the Selkies desktop streaming balance. Resolution remains adaptive to the browser window.",
+      desktopStreamLow: "Low",
+      desktopStreamLowDesc:
+        "Best for weak networks and large concurrency. Keeps motion usable while reducing clarity.",
+      desktopStreamStandard: "Standard",
+      desktopStreamStandardDesc:
+        "Balanced default with responsive motion and moderate clarity.",
+      desktopStreamHigh: "High",
+      desktopStreamHighDesc:
+        "Smoother motion with higher bandwidth and CPU usage.",
+      restartRequiredAfterChange:
+        "Changes take effect after the instance is restarted.",
+      desktopStreamSavedRestartRequired:
+        "Stream profile saved. Restart the instance for it to take effect.",
+      desktopStreamSaveFailed: "Failed to save stream profile.",
       environmentVariables: "Umgebungsvariablen",
       clawManagerBuiltIns: "ClawManager Built-ins",
       hideBuiltIns: "Built-ins ausblenden",

frontend/src/pages/instances/CreateInstancePage.tsx

@@ -20,6 +20,7 @@ import {
   systemSettingsService,
   type SystemImageSetting,
 } from "../../services/systemSettingsService";
+import type { DesktopStreamProfile } from "../../types/instance";
 
 type BuiltInEnvTemplate = {
   key: string;
@@ -46,6 +47,31 @@ const CUSTOM_RESOURCE_PRESET = "custom";
 const SKILLS_PER_PAGE = 6;
 const supportsRuntimeInjection = (type: string) =>
   type === "openclaw" || type === "hermes";
+const DESKTOP_STREAM_PROFILES: Array<{
+  id: DesktopStreamProfile;
+  labelKey: string;
+  descriptionKey: string;
+  details: string;
+}> = [
+  {
+    id: "low",
+    labelKey: "instances.desktopStreamLow",
+    descriptionKey: "instances.desktopStreamLowDesc",
+    details: "30 FPS / CRF 42",
+  },
+  {
+    id: "standard",
+    labelKey: "instances.desktopStreamStandard",
+    descriptionKey: "instances.desktopStreamStandardDesc",
+    details: "35 FPS / CRF 34",
+  },
+  {
+    id: "high",
+    labelKey: "instances.desktopStreamHigh",
+    descriptionKey: "instances.desktopStreamHighDesc",
+    details: "40 FPS / CRF 24",
+  },
+];
 
 const runtimeWorkspaceDirectory = (type: string) =>
   type === "hermes" ? ".hermes" : ".openclaw";
@@ -442,6 +468,8 @@ const CreateInstancePage: React.FC = () => {
     Record<string, string>
   >({});
   const [customEnvRows, setCustomEnvRows] = useState<CustomEnvRow[]>([]);
+  const [desktopStreamProfile, setDesktopStreamProfile] =
+    useState<DesktopStreamProfile>("standard");
   const [resourcePresetMode, setResourcePresetMode] = useState<
     keyof typeof PRESET_CONFIGS | typeof CUSTOM_RESOURCE_PRESET
   >("medium");
@@ -821,6 +849,8 @@ const CreateInstancePage: React.FC = () => {
           : PRESET_CONFIGS.small.disk_gb,
         gpu_enabled: usesDedicatedResources ? formData.gpu_enabled : false,
         gpu_count: usesDedicatedResources ? formData.gpu_count : 0,
+        desktop_stream_profile:
+          selectedRuntimeType === "desktop" ? desktopStreamProfile : undefined,
         image_registry: selectedRuntimeImage?.image,
         image_tag: selectedRuntimeImage ? undefined : formData.image_tag,
         environment_overrides: overrides,
@@ -1579,6 +1609,54 @@ const CreateInstancePage: React.FC = () => {
                   </div>
                 )}
 
+                {selectedRuntimeType === "desktop" && (
+                  <div className="mt-6 rounded-[24px] border border-gray-200 bg-gray-50/80 p-4">
+                    <div className="flex flex-col gap-2 sm:flex-row sm:items-start sm:justify-between">
+                      <div>
+                        <h3 className="text-sm font-semibold uppercase tracking-[0.2em] text-gray-500">
+                          {t("instances.desktopStreamProfile")}
+                        </h3>
+                        <p className="mt-2 text-sm text-gray-500">
+                          {t("instances.desktopStreamProfileDesc")}
+                        </p>
+                      </div>
+                      <span className="rounded-full bg-white px-3 py-1 text-xs font-medium text-gray-500">
+                        {t("instances.restartRequiredAfterChange")}
+                      </span>
+                    </div>
+
+                    <div className="mt-4 grid grid-cols-1 gap-3 md:grid-cols-3">
+                      {DESKTOP_STREAM_PROFILES.map((profile) => {
+                        const selected = desktopStreamProfile === profile.id;
+                        return (
+                          <button
+                            key={profile.id}
+                            type="button"
+                            onClick={() => setDesktopStreamProfile(profile.id)}
+                            className={`flex min-h-[138px] flex-col justify-between rounded-[20px] border p-4 text-left transition ${
+                              selected
+                                ? "border-indigo-500 bg-white ring-2 ring-indigo-500"
+                                : "border-gray-200 bg-white hover:border-indigo-200"
+                            }`}
+                          >
+                            <div>
+                              <h4 className="text-sm font-semibold text-gray-900">
+                                {t(profile.labelKey)}
+                              </h4>
+                              <p className="mt-2 text-sm leading-5 text-gray-500">
+                                {t(profile.descriptionKey)}
+                              </p>
+                            </div>
+                            <p className="mt-4 font-mono text-xs text-gray-400">
+                              {profile.details}
+                            </p>
+                          </button>
+                        );
+                      })}
+                    </div>
+                  </div>
+                )}
+
                 <div className="app-panel order-3 p-6">
                   <div className="flex items-start justify-between gap-4">
                     <div>

frontend/src/pages/instances/InstanceDetailPage.tsx

@@ -40,11 +40,21 @@ import type {
   InstanceRuntimeCommand,
   InstanceRuntimeDetails,
   InstanceStatus,
+  DesktopStreamProfile,
 } from "../../types/instance";
 import type { InstanceSkill, Skill } from "../../types/skill";
 
 const META_POLL_INTERVAL_MS = 5000;
 const RUNTIME_POLL_INTERVAL_MS = 5000;
+const DESKTOP_STREAM_PROFILES: Array<{
+  id: DesktopStreamProfile;
+  labelKey: string;
+  detail: string;
+}> = [
+  { id: "low", labelKey: "instances.desktopStreamLow", detail: "30 FPS / CRF 42" },
+  { id: "standard", labelKey: "instances.desktopStreamStandard", detail: "35 FPS / CRF 34" },
+  { id: "high", labelKey: "instances.desktopStreamHigh", detail: "40 FPS / CRF 24" },
+];
 
 function availabilityForStatus(status: string): InstanceAvailability {
   if (status === "running") {
@@ -304,6 +314,11 @@ const InstanceDetailPage: React.FC = () => {
   const [selectedSkillId, setSelectedSkillId] = useState<number | "">("");
   const [skillLoading, setSkillLoading] = useState(false);
   const [skillError, setSkillError] = useState<string | null>(null);
+  const [desktopStreamProfile, setDesktopStreamProfile] =
+    useState<DesktopStreamProfile>("standard");
+  const [desktopStreamSavedProfile, setDesktopStreamSavedProfile] =
+    useState<DesktopStreamProfile>("standard");
+  const [desktopStreamMessage, setDesktopStreamMessage] = useState<string | null>(null);
 
   const fetchMeta = useCallback(
     async (targetInstanceId: number, options?: { background?: boolean }) => {
@@ -379,6 +394,13 @@ const InstanceDetailPage: React.FC = () => {
     }
   }, []);
 
+  useEffect(() => {
+    const profile = instance?.desktop_stream_profile || "standard";
+    setDesktopStreamProfile(profile);
+    setDesktopStreamSavedProfile(profile);
+    setDesktopStreamMessage(null);
+  }, [instance?.desktop_stream_profile, instance?.id]);
+
   useEffect(() => {
     if (!instanceId || Number.isNaN(instanceId)) {
       setError(t("instances.instanceNotFound"));
@@ -562,6 +584,31 @@ const InstanceDetailPage: React.FC = () => {
     }
   };
 
+  const handleSaveDesktopStreamProfile = async () => {
+    if (!instance || desktopStreamProfile === desktopStreamSavedProfile) {
+      return;
+    }
+
+    try {
+      setActionLoading("desktop-stream-profile");
+      setDesktopStreamMessage(null);
+      await instanceService.updateInstance(instance.id, {
+        desktop_stream_profile: desktopStreamProfile,
+      });
+      const updatedInstance = await instanceService.getInstance(instance.id);
+      setInstance(updatedInstance);
+      const savedProfile = updatedInstance.desktop_stream_profile || desktopStreamProfile;
+      setDesktopStreamSavedProfile(savedProfile);
+      setDesktopStreamProfile(savedProfile);
+      setDesktopStreamMessage(t("instances.desktopStreamSavedRestartRequired"));
+    } catch (streamError) {
+      console.error("Failed to update desktop stream profile", streamError);
+      setDesktopStreamMessage(t("instances.desktopStreamSaveFailed"));
+    } finally {
+      setActionLoading(null);
+    }
+  };
+
   const copyExternalValue = async (key: string, value: string) => {
     if (!value) {
       return;
@@ -983,6 +1030,55 @@ const InstanceDetailPage: React.FC = () => {
         )}
       </section>
 
+      {(instance.runtime_type || "desktop") === "desktop" && (
+        <section className="cm-surface px-4 py-4">
+          <div className="mb-3 flex flex-col gap-3 sm:flex-row sm:items-start sm:justify-between">
+            <div>
+              <h2 className="text-sm font-semibold text-slate-950">
+                {t("instances.desktopStreamProfile")}
+              </h2>
+              <p className="mt-1 text-xs text-slate-500">
+                {t("instances.desktopStreamProfileDesc")}
+              </p>
+            </div>
+            <button
+              type="button"
+              onClick={handleSaveDesktopStreamProfile}
+              disabled={
+                actionLoading === "desktop-stream-profile" ||
+                desktopStreamProfile === desktopStreamSavedProfile
+              }
+              className="app-button-secondary shrink-0 disabled:cursor-not-allowed disabled:opacity-50"
+            >
+              {actionLoading === "desktop-stream-profile" ? t("common.saving") : t("common.save")}
+            </button>
+          </div>
+          <div className="grid gap-2 sm:grid-cols-3">
+            {DESKTOP_STREAM_PROFILES.map((profile) => {
+              const selected = desktopStreamProfile === profile.id;
+              return (
+                <button
+                  key={profile.id}
+                  type="button"
+                  onClick={() => setDesktopStreamProfile(profile.id)}
+                  className={`min-h-[82px] rounded-md border px-3 py-2 text-left transition ${
+                    selected
+                      ? "border-indigo-500 bg-indigo-50 ring-2 ring-indigo-100"
+                      : "border-slate-200 bg-white hover:border-indigo-200"
+                  }`}
+                >
+                  <div className="text-sm font-semibold text-slate-950">{t(profile.labelKey)}</div>
+                  <div className="mt-2 font-mono text-xs text-slate-500">{profile.detail}</div>
+                </button>
+              );
+            })}
+          </div>
+          <p className="mt-3 text-xs text-slate-500">
+            {desktopStreamMessage || t("instances.restartRequiredAfterChange")}
+          </p>
+        </section>
+      )}
+
       <section className="grid gap-4 xl:grid-cols-[minmax(0,1fr)_minmax(280px,22rem)]">
         <section className="cm-surface px-4 py-4">
           <div className="mb-3 flex items-center justify-between gap-3">

frontend/src/services/instanceService.ts

@@ -132,6 +132,8 @@ export const instanceService = {
     access_url: string;
     proxy_url: string;
     expires_at: string;
+    desktop_proxy_mode?: "control-plane" | "fallback" | "direct";
+    desktop_upstream_present?: boolean;
   }> => {
     const response = await api.post(`/instances/${id}/access`);
     return response.data.data;

frontend/src/types/instance.ts

@@ -26,6 +26,7 @@ export interface Instance {
   os_version: string;
   image_registry?: string;
   image_tag?: string;
+  desktop_stream_profile?: DesktopStreamProfile;
   storage_class: string;
   mount_path: string;
   workspace_path?: string;
@@ -179,6 +180,7 @@ export interface CreateInstanceRequest {
   mode?: InstanceMode;
   instance_mode?: InstanceMode;
   runtime_type?: "desktop" | "shell" | "gateway";
+  desktop_stream_profile?: DesktopStreamProfile;
   cpu_cores: number;
   memory_gb: number;
   disk_gb: number;
@@ -197,8 +199,11 @@ export interface CreateInstanceRequest {
 export interface UpdateInstanceRequest {
   name?: string;
   description?: string;
+  desktop_stream_profile?: DesktopStreamProfile;
 }
 
+export type DesktopStreamProfile = "low" | "standard" | "high";
+
 export interface InstanceListResponse {
   instances: Instance[];
   total: number;