feat: Add option to disable cluster websocket connection and fix instability (#735)

maciaszczykm · greptile-apps[bot] · web-flow · commit ec69b5dea714 · 2026-05-08T16:39:48.000+02:00
* feat: Add option to disable cluster websocket connection (#684) * add option to disable cluster websocket connection * add option to disable websocket connection in console manager * remove websocket configuration check and simplify console manager setup * simplify console manager initialization by removing unnecessary parameters * streamline socket connection handling in controller manager * add Close method to Socket interface and implement in socket * add option to disable websocket connection in controller manager * disable websocket in agent configuration * make socket methods thread-safe by adding mutex locks * refactor startController method to improve readability * add reconnect logic to websocket with `closed` state handling and improve thread safety * improve comments in socket.go for clarity and consistency * refactor websocket connection handling to improve clarity and prevent blocking * refactor wssUri function to construct websocket URL more clearly and efficiently * remove redundant comments * enhance websocket connection handling with improved error messages and initialization * refactor websocket event handling to improve clarity, error messaging, and validation * introduce `joining` state to websocket for improved thread safety and prevent deadlock, and enable websocket in agent configuration * refactor websocket reconnect logic for clarity, removing redundant `reconnect` method * fix socket message handling to ensure proper mutex unlocking when socket is closed * refactor * adjust log levels * ensure proper state management on connection failure * ensure proper state management on connection failure * update websocket disconnection logic * enhance websocket disconnection handling with reconnect logic * update websocket disconnection message and handle client closure * add option to create new websocket client with generation tracking * add clientReceiver to manage websocket client generation and prevent stale callbacks * refactor notifyConnectLocked to return a boolean for Join() invocation * Update pkg/websocket/client_receiver.go Co-authored-by: greptile-apps[bot] <165735046+greptile-apps[bot]@users.noreply.github.com> --------- Co-authored-by: greptile-apps[bot] <165735046+greptile-apps[bot]@users.noreply.github.com> * avoid callback-driven client teardown on transient disconnects and let gophoenix own transport reconnects to prevent protocol error churn * extend tests * fix review finding * initialize sockets in closed state to avoid startup websocket handshakes * fix lint * generate crd docs * fix review findings * fix review findings * switch to graphql subscriptions * add logs * add log * normalize resource value * fix review findings * improve error handling in AgentRuntimeReconciler * add logs * update publisher key * update other keys --------- Co-authored-by: greptile-apps[bot] <165735046+greptile-apps[bot]@users.noreply.github.com>
diff --git a/api/v1alpha1/agentconfig_types.go b/api/v1alpha1/agentconfig_types.go
@@ -54,6 +54,12 @@ type AgentConfigurationSpec struct {
 	// Must be greater than 0. Set this field to nil (omit) to disable the limit.
 	// +kubebuilder:validation:Minimum=1
 	MaxAgentRunPods *int `json:"maxAgentRunPods,omitempty"`
+
+	// DisableWebsocket disables the cluster websocket connection to the Console.
+	// When enabled, the agent will rely exclusively on polling instead of receiving
+	// push updates. This is useful in large-scale edge deployments where maintaining
+	// persistent websocket connections has an infeasible network cost.
+	DisableWebsocket *bool `json:"disableWebsocket,omitempty"`
 }
 
 //+kubebuilder:object:root=true
diff --git a/api/v1alpha1/zz_generated.deepcopy.go b/api/v1alpha1/zz_generated.deepcopy.go
diff --git a/charts/deployment-operator/crds/deployments.plural.sh_agentconfigurations.yaml b/charts/deployment-operator/crds/deployments.plural.sh_agentconfigurations.yaml
@@ -60,6 +60,13 @@ spec:
                   Must be greater than 0. Set this field to nil (omit) to disable the limit.
                 minimum: 1
                 type: integer
+              disableWebsocket:
+                description: |-
+                  DisableWebsocket disables the cluster websocket connection to the Console.
+                  When enabled, the agent will rely exclusively on polling instead of receiving
+                  push updates. This is useful in large-scale edge deployments where maintaining
+                  persistent websocket connections has an infeasible network cost.
+                type: boolean
               maxConcurrentReconciles:
                 description: |-
                   MaxConcurrentReconciles controls the maximum number of concurrent reconcile loops.
diff --git a/cmd/agent/kubernetes.go b/cmd/agent/kubernetes.go
@@ -317,7 +317,7 @@ func registerKubeReconcilersOrDie(
 	if err := agentRuntimeReconciler.SetupWithManager(manager); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "AgentRuntime")
 	}
-	consoleManager.Socket.AddPublisher("agent.run.event", agentRuntimeReconciler)
+	consoleManager.Socket.AddPublisher("agent_run", agentRuntimeReconciler)
 
 	if err := (&controller.AgentRunReconciler{
 		Client:           manager.GetClient(),
diff --git a/config/crd/bases/deployments.plural.sh_agentconfigurations.yaml b/config/crd/bases/deployments.plural.sh_agentconfigurations.yaml
@@ -60,6 +60,13 @@ spec:
                   Must be greater than 0. Set this field to nil (omit) to disable the limit.
                 minimum: 1
                 type: integer
+              disableWebsocket:
+                description: |-
+                  DisableWebsocket disables the cluster websocket connection to the Console.
+                  When enabled, the agent will rely exclusively on polling instead of receiving
+                  push updates. This is useful in large-scale edge deployments where maintaining
+                  persistent websocket connections has an infeasible network cost.
+                type: boolean
               maxConcurrentReconciles:
                 description: |-
                   MaxConcurrentReconciles controls the maximum number of concurrent reconcile loops.
diff --git a/config/samples/agentConfiguration.yaml b/config/samples/agentConfiguration.yaml
@@ -12,4 +12,5 @@ spec:
   vulnerabilityReportUploadInterval: "10m"
   maxConcurrentReconciles: 5
   maxSentinelRunJobs: 2
+  disableWebsocket: false
 
diff --git a/docs/api.md b/docs/api.md
@@ -86,6 +86,7 @@ _Appears in:_
 | `maxSentinelRunJobs` _integer_ | MaxSentinelRunJobs limits the number of concurrent SentinelRunJobs that can be active at any given time.<br />Must be greater than 0. Set this field to nil (omit) to disable the limit. |  | Minimum: 1 <br /> |
 | `maxStackRunJobs` _integer_ | MaxStackRunJobs limits the number of concurrent StackRunJobs that can be active at any given time.<br />Must be greater than 0. Set this field to nil (omit) to disable the limit. |  | Minimum: 1 <br /> |
 | `maxAgentRunPods` _integer_ | MaxAgentRunPods limits the number of concurrent agent run pods that can be active at any given time.<br />Must be greater than 0. Set this field to nil (omit) to disable the limit. |  | Minimum: 1 <br /> |
+| `disableWebsocket` _boolean_ | DisableWebsocket disables the cluster websocket connection to the Console.<br />When enabled, the agent will rely exclusively on polling instead of receiving<br />push updates. This is useful in large-scale edge deployments where maintaining<br />persistent websocket connections has an infeasible network cost. |  |  |
 
 
 #### AgentHelmConfiguration
diff --git a/go.mod b/go.mod
@@ -30,6 +30,7 @@ require (
 	github.com/google/go-github/v68 v68.0.0
 	github.com/grafana/pyroscope-go v1.2.7
 	github.com/hashicorp/terraform-json v0.26.0
+	github.com/hasura/go-graphql-client v0.16.0
 	github.com/mark3labs/mcp-go v0.40.0
 	github.com/mitchellh/copystructure v1.2.0
 	github.com/onsi/ginkgo/v2 v2.28.1
@@ -44,7 +45,6 @@ require (
 	github.com/pluralsh/console/go/client v1.74.1
 	github.com/pluralsh/console/go/polly v1.0.0
 	github.com/pluralsh/controller-reconcile-helper v0.1.0
-	github.com/pluralsh/gophoenix v0.1.3-0.20231201014135-dff1b4309e34
 	github.com/prometheus/client_golang v1.23.2
 	github.com/sahilm/fuzzy v0.1.1
 	github.com/samber/lo v1.52.0
@@ -193,6 +193,7 @@ require (
 	github.com/ckaznocha/intrange v0.3.1 // indirect
 	github.com/cloudwego/base64x v0.1.4 // indirect
 	github.com/cloudwego/iasm v0.2.0 // indirect
+	github.com/coder/websocket v1.8.14 // indirect
 	github.com/containerd/containerd v1.7.30 // indirect
 	github.com/containerd/errdefs v1.0.0 // indirect
 	github.com/containerd/log v0.1.0 // indirect
@@ -282,7 +283,6 @@ require (
 	github.com/google/shlex v0.0.0-20191202100458-e7afc7fbc510 // indirect
 	github.com/google/uuid v1.6.0 // indirect
 	github.com/gordonklaus/ineffassign v0.2.0 // indirect
-	github.com/gorilla/websocket v1.5.4-0.20250319132907-e064f32e3674 // indirect
 	github.com/gostaticanalysis/analysisutil v0.7.1 // indirect
 	github.com/gostaticanalysis/comment v1.5.0 // indirect
 	github.com/gostaticanalysis/forcetypeassert v0.2.0 // indirect
@@ -311,11 +311,9 @@ require (
 	github.com/jjti/go-spancheck v0.6.5 // indirect
 	github.com/jmoiron/sqlx v1.4.0 // indirect
 	github.com/josharian/intern v1.0.0 // indirect
-	github.com/jpillora/backoff v1.0.0 // indirect
 	github.com/json-iterator/go v1.1.12 // indirect
 	github.com/julz/importas v0.2.0 // indirect
 	github.com/karamaru-alpha/copyloopvar v1.2.2 // indirect
-	github.com/kataras/golog v0.1.15 // indirect
 	github.com/kisielk/errcheck v1.10.0 // indirect
 	github.com/kkHAIKE/contextcheck v1.1.6 // indirect
 	github.com/klauspost/compress v1.18.2 // indirect
diff --git a/go.sum b/go.sum
@@ -339,6 +339,8 @@ github.com/cloudwego/iasm v0.2.0 h1:1KNIy1I1H9hNNFEEH3DVnI4UujN+1zjpuk6gwHLTssg=
 github.com/cloudwego/iasm v0.2.0/go.mod h1:8rXZaNYT2n95jn+zTI1sDr+IgcD2GVs0nlbbQPiEFhY=
 github.com/cncf/xds/go v0.0.0-20251210132809-ee656c7534f5 h1:6xNmx7iTtyBRev0+D/Tv1FZd4SCg8axKApyNyRsAt/w=
 github.com/cncf/xds/go v0.0.0-20251210132809-ee656c7534f5/go.mod h1:KdCmV+x/BuvyMxRnYBlmVaq4OLiKW6iRQfvC62cvdkI=
+github.com/coder/websocket v1.8.14 h1:9L0p0iKiNOibykf283eHkKUHHrpG7f65OE3BhhO7v9g=
+github.com/coder/websocket v1.8.14/go.mod h1:NX3SzP+inril6yawo5CQXx8+fk145lPDC6pumgx0mVg=
 github.com/containerd/containerd v1.7.29 h1:90fWABQsaN9mJhGkoVnuzEY+o1XDPbg9BTC9QTAHnuE=
 github.com/containerd/containerd v1.7.29/go.mod h1:azUkWcOvHrWvaiUjSQH0fjzuHIwSPg1WL5PshGP4Szs=
 github.com/containerd/errdefs v1.0.0 h1:tg5yIfIlQIrxYtu9ajqY42W3lpS19XqdxRQeEwYG8PI=
@@ -640,8 +642,6 @@ github.com/gorilla/handlers v1.5.2 h1:cLTUSsNkgcwhgRqvCNmdbRWG0A3N4F+M2nWKdScwyE
 github.com/gorilla/handlers v1.5.2/go.mod h1:dX+xVpaxdSw+q0Qek8SSsl3dfMk3jNddUkMzo0GtH0w=
 github.com/gorilla/mux v1.8.1 h1:TuBL49tXwgrFYWhqrNgrUNEY92u81SPhu7sTdzQEiWY=
 github.com/gorilla/mux v1.8.1/go.mod h1:AKf9I4AEqPTmMytcMc0KkNouC66V3BtZ4qD5fmWSiMQ=
-github.com/gorilla/websocket v1.5.4-0.20250319132907-e064f32e3674 h1:JeSE6pjso5THxAzdVpqr6/geYxZytqFMBCOtn/ujyeo=
-github.com/gorilla/websocket v1.5.4-0.20250319132907-e064f32e3674/go.mod h1:r4w70xmWCQKmi1ONH4KIaBptdivuRPyosB9RmPlGEwA=
 github.com/gostaticanalysis/analysisutil v0.7.1 h1:ZMCjoue3DtDWQ5WyU16YbjbQEQ3VuzwxALrpYd+HeKk=
 github.com/gostaticanalysis/analysisutil v0.7.1/go.mod h1:v21E3hY37WKMGSnbsw2S/ojApNWb6C1//mXO48CXbVc=
 github.com/gostaticanalysis/comment v1.4.2/go.mod h1:KLUTGDv6HOCotCH8h2erHKmpci2ZoR8VPu34YA2uzdM=
@@ -699,6 +699,8 @@ github.com/hashicorp/hcl/v2 v2.24.0 h1:2QJdZ454DSsYGoaE6QheQZjtKZSUs9Nh2izTWiwQx
 github.com/hashicorp/hcl/v2 v2.24.0/go.mod h1:oGoO1FIQYfn/AgyOhlg9qLC6/nOJPX3qGbkZpYAcqfM=
 github.com/hashicorp/terraform-json v0.26.0 h1:+BnJavhRH+oyNWPnfzrfQwVWCZBFMvjdiH2Vi38Udz4=
 github.com/hashicorp/terraform-json v0.26.0/go.mod h1:eyWCeC3nrZamyrKLFnrvwpc3LQPIJsx8hWHQ/nu2/v4=
+github.com/hasura/go-graphql-client v0.16.0 h1:DQLfp+djj4j5NPdJkGYym8J55hpm5etML1zqgco78Qc=
+github.com/hasura/go-graphql-client v0.16.0/go.mod h1:z/sO2T0zI+HnPNIevQcs+7xA6/gDOc8hgHMrNBzfL2c=
 github.com/hexops/gotextdiff v1.0.3 h1:gitA9+qJrrTCsiCl7+kh75nPqQt1cx4ZkudSTLoUqJM=
 github.com/hexops/gotextdiff v1.0.3/go.mod h1:pSWU5MAI3yDq+fZBTazCSJysOMbxWL1BSow5/V2vxeg=
 github.com/huandu/xstrings v1.5.0 h1:2ag3IFq9ZDANvthTwTiqSSZLjDc+BedvHPAp5tJy2TI=
@@ -727,16 +729,12 @@ github.com/josharian/intern v1.0.0 h1:vlS4z54oSdjm0bgjRigI+G1HpF+tI+9rE5LLzOg8Hm
 github.com/josharian/intern v1.0.0/go.mod h1:5DoeVV0s6jJacbCEi61lwdGj/aVlrQvzHFFd8Hwg//Y=
 github.com/joshdk/go-junit v1.0.0 h1:S86cUKIdwBHWwA6xCmFlf3RTLfVXYQfvanM5Uh+K6GE=
 github.com/joshdk/go-junit v1.0.0/go.mod h1:TiiV0PqkaNfFXjEiyjWM3XXrhVyCa1K4Zfga6W52ung=
-github.com/jpillora/backoff v1.0.0 h1:uvFg412JmmHBHw7iwprIxkPMI+sGQ4kzOWsMeHnm2EA=
-github.com/jpillora/backoff v1.0.0/go.mod h1:J/6gKK9jxlEcS3zixgDgUAsiuZ7yrSoa/FX5e0EB2j4=
 github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM=
 github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo=
 github.com/julz/importas v0.2.0 h1:y+MJN/UdL63QbFJHws9BVC5RpA2iq0kpjrFajTGivjQ=
 github.com/julz/importas v0.2.0/go.mod h1:pThlt589EnCYtMnmhmRYY/qn9lCf/frPOK+WMx3xiJY=
 github.com/karamaru-alpha/copyloopvar v1.2.2 h1:yfNQvP9YaGQR7VaWLYcfZUlRP2eo2vhExWKxD/fP6q0=
 github.com/karamaru-alpha/copyloopvar v1.2.2/go.mod h1:oY4rGZqZ879JkJMtX3RRkcXRkmUvH0x35ykgaKgsgJY=
-github.com/kataras/golog v0.1.15 h1:gDNOENbbn+6me98UW1f9Cs5MRUlAkabnNvmgLFM58Xw=
-github.com/kataras/golog v0.1.15/go.mod h1:Ozu1TDa+OKC7fFe7OG64In71yLxjda+6kPl+Rg3v1hA=
 github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8=
 github.com/kisielk/errcheck v1.10.0 h1:Lvs/YAHP24YKg08LA8oDw2z9fJVme090RAXd90S+rrw=
 github.com/kisielk/errcheck v1.10.0/go.mod h1:kQxWMMVZgIkDq7U8xtG/n2juOjbLgZtedi0D+/VL/i8=
@@ -947,8 +945,6 @@ github.com/pluralsh/console/go/polly v1.0.0 h1:NRg8CMITGllEJ08oYidNuuG/gJGDdDuga
 github.com/pluralsh/console/go/polly v1.0.0/go.mod h1:eH42iPlRip2bs6tB+Vg6fxC/sz6oWxyNGVDE9uyA5EY=
 github.com/pluralsh/controller-reconcile-helper v0.1.0 h1:BV3dYZFH5rn8ZvZjtpkACSv/GmLEtRftNQj/Y4ddHEo=
 github.com/pluralsh/controller-reconcile-helper v0.1.0/go.mod h1:RxAbvSB4/jkvx616krCdNQXPbpGJXW3J1L3rASxeFOA=
-github.com/pluralsh/gophoenix v0.1.3-0.20231201014135-dff1b4309e34 h1:ab2PN+6if/Aq3/sJM0AVdy1SYuMAnq4g20VaKhTm/Bw=
-github.com/pluralsh/gophoenix v0.1.3-0.20231201014135-dff1b4309e34/go.mod h1:IagWXKFYu6NTHzcJx2dJyrIlZ1Sv2PH3fhOtplA9qOs=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRIccs7FGNTlIRMkT8wgtp5eCXdBlqhYGL6U=
 github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
diff --git a/internal/controller/agentruntime_controller.go b/internal/controller/agentruntime_controller.go
@@ -212,42 +212,46 @@ func (r *AgentRuntimeReconciler) addOrRemoveFinalizer(ctx context.Context, agent
 	return nil
 }
 
-func (r *AgentRuntimeReconciler) Publish(runId string, kick bool) {
-	err := r.createRunFromId(runId)
+func (r *AgentRuntimeReconciler) Publish(runID string, kick bool) {
+	logger := log.FromContext(r.Ctx)
+	logger.Info("received websocket agent run event", "id", runID, "kick", kick)
+
+	err := r.createRunFromID(runID)
 	if err != nil {
-		log := log.FromContext(r.Ctx)
-		log.Error(err, "failed to create agent run", "id", runId)
+		logger.Error(err, "failed to process websocket agent run event", "id", runID, "kick", kick)
 	}
 }
 
-func (r *AgentRuntimeReconciler) createRunFromId(runId string) error {
-	ctx := r.Ctx
-
-	run, err := r.ConsoleClient.GetAgentRun(ctx, runId)
+func (r *AgentRuntimeReconciler) createRunFromID(runID string) error {
+	logger := log.FromContext(r.Ctx)
+	run, err := r.ConsoleClient.GetAgentRun(r.Ctx, runID)
 	if err != nil {
 		return fmt.Errorf("failed to get agent run: %w", err)
 	}
-
-	runtimeName := run.Runtime.Name
+	if run.Runtime == nil || run.Runtime.Name == "" {
+		return fmt.Errorf("agent run %q runtime details are missing", runID)
+	}
 
 	agentRuntime := &v1alpha1.AgentRuntime{}
-	nsn := types.NamespacedName{
-		Name:      runtimeName,
-		Namespace: "",
-	}
-	if err := r.Get(ctx, nsn, agentRuntime); err != nil {
+	if err := r.Get(r.Ctx, types.NamespacedName{Name: run.Runtime.Name, Namespace: ""}, agentRuntime); err != nil {
+		if errors.IsNotFound(err) {
+			logger.Info("agent runtime for websocket agent run event not found", "id", runID, "runtime", run.Runtime.Name)
+		}
 		return client.IgnoreNotFound(err)
 	}
 
-	return r.createAgentRun(ctx, agentRuntime, run)
+	return r.createAgentRun(r.Ctx, agentRuntime, run)
 }
 
 func (r *AgentRuntimeReconciler) createAgentRun(ctx context.Context, agentRuntime *v1alpha1.AgentRuntime, run *console.AgentRunFragment) error {
 	logger := log.FromContext(ctx)
+	key := client.ObjectKey{Name: run.ID, Namespace: agentRuntime.Spec.TargetNamespace}
 
-	if r.Get(ctx, client.ObjectKey{Name: run.ID, Namespace: agentRuntime.Spec.TargetNamespace}, &v1alpha1.AgentRun{}) == nil {
+	if err := r.Get(ctx, key, &v1alpha1.AgentRun{}); err == nil {
 		logger.Info("agent run already exists", "id", run.ID)
 		return nil
+	} else if !errors.IsNotFound(err) {
+		return fmt.Errorf("failed to lookup agent run: %w", err)
 	}
 
 	agentRun := &v1alpha1.AgentRun{
@@ -277,6 +281,9 @@ func (r *AgentRuntimeReconciler) createAgentRun(ctx context.Context, agentRuntim
 	}
 
 	if err := r.Create(ctx, agentRun); err != nil {
+		if errors.IsAlreadyExists(err) {
+			return nil
+		}
 		return fmt.Errorf("failed to create agent run: %w", err)
 	}
 
diff --git a/pkg/common/config.go b/pkg/common/config.go
@@ -36,6 +36,7 @@ type ConfigurationManager struct {
 	maxStackRunJobs             *int
 	maxAgentRunPods             *int
 	baseRegistryURL             *string
+	disableWebsocket            *bool
 }
 
 func GetConfigurationManager() *ConfigurationManager {
@@ -86,6 +87,7 @@ func (s *ConfigurationManager) SetValue(config v1alpha1.AgentConfigurationSpec)
 	s.maxConcurrentReconciles = config.MaxConcurrentReconciles
 	s.baseRegistryURL = config.BaseRegistryURL
 	s.maxSentinelRunJobs = config.MaxSentinelRunJobs
+	s.disableWebsocket = config.DisableWebsocket
 
 	return nil
 }
@@ -176,6 +178,12 @@ func (s *ConfigurationManager) GetBaseRegistryURL() *string {
 	return s.baseRegistryURL
 }
 
+func (s *ConfigurationManager) IsWebsocketDisabled() bool {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	return s.disableWebsocket != nil && *s.disableWebsocket
+}
+
 func (s *ConfigurationManager) SwapBaseRegistry(image string) string {
 	s.mu.RLock()
 	defer s.mu.RUnlock()
diff --git a/pkg/controller/controller_manager.go b/pkg/controller/controller_manager.go
@@ -13,6 +13,7 @@ import (
 	"github.com/pluralsh/deployment-operator/internal/helpers"
 	"github.com/pluralsh/deployment-operator/internal/metrics"
 	"github.com/pluralsh/deployment-operator/pkg/client"
+	"github.com/pluralsh/deployment-operator/pkg/common"
 	"github.com/pluralsh/deployment-operator/pkg/controller/service"
 	v1 "github.com/pluralsh/deployment-operator/pkg/controller/v1"
 	"github.com/pluralsh/deployment-operator/pkg/log"
@@ -106,6 +107,13 @@ func (cm *Manager) Start(ctx context.Context) error {
 	go cm.startSupervised(ctx)
 
 	_ = helpers.DynamicBackgroundPollUntilContextCancel(ctx, func() time.Duration { return cm.PollInterval }, false, func(_ context.Context) (bool, error) {
+		if common.GetConfigurationManager().IsWebsocketDisabled() {
+			if err := cm.Socket.Close(); err != nil {
+				klog.ErrorS(err, "unable to close websocket")
+			}
+			return false, nil
+		}
+
 		if err := cm.Socket.Join(); err != nil {
 			klog.ErrorS(err, "unable to connect")
 		}
@@ -238,9 +246,7 @@ func (cm *Manager) startControllerSupervised(ctx context.Context, ctrl *Controll
 // startController starts the controller and blocks until it does not stop.
 func (cm *Manager) startController(ctx context.Context, ctrl *Controller) {
 	klog.V(log.LogLevelDefault).InfoS("Starting controller", "name", ctrl.Name)
-
-	// If publisher exists, this is a no-op
-	cm.Socket.AddPublisher(ctrl.Do.GetPublisher())
+	cm.Socket.AddPublisher(ctrl.Do.GetPublisher()) // If publisher exists, this is a no-op
 	ctrl.Start(ctx)
 }
 
diff --git a/pkg/controller/controller_manager_options.go b/pkg/controller/controller_manager_options.go
@@ -3,8 +3,6 @@ package controller
 import (
 	"time"
 
-	"k8s.io/klog/v2"
-
 	"github.com/pluralsh/deployment-operator/pkg/client"
 	"github.com/pluralsh/deployment-operator/pkg/websocket"
 )
@@ -34,14 +32,11 @@ func WithSocket(socket websocket.Socket) ControllerManagerOption {
 
 func WithSocketArgs(clusterID, url, deployToken string) ControllerManagerOption {
 	return func(o *Manager) (err error) {
-		socket, err := websocket.New(clusterID, url, deployToken)
+		// Start with a closed socket and let manager poll decide when to open it.
+		// This avoids establishing a websocket connection before runtime config
+		// (disableWebsocket) has been applied by AgentConfiguration reconciliation.
+		socket, err := websocket.NewClosed(clusterID, url, deployToken)
 		o.Socket = socket
-
-		if err != nil && socket != nil {
-			klog.Error(err, "could not initiate websocket connection, ignoring and falling back to polling")
-			return nil
-		}
-
 		return err
 	}
 }
diff --git a/pkg/controller/namespaces/reconciler.go b/pkg/controller/namespaces/reconciler.go
@@ -72,7 +72,7 @@ func (n *NamespaceReconciler) GetPollInterval() func() time.Duration {
 }
 
 func (n *NamespaceReconciler) GetPublisher() (string, websocket.Publisher) {
-	return "namespace.event", &socketPublisher{
+	return "namespace", &socketPublisher{
 		restoreQueue: n.namespaceQueue,
 		restoreCache: n.namespaceCache,
 	}
diff --git a/pkg/controller/pipelinegates/reconciler.go b/pkg/controller/pipelinegates/reconciler.go
@@ -163,7 +163,7 @@ func (s *GateReconciler) Reconcile(ctx context.Context, id string) (reconcile.Re
 }
 
 func (s *GateReconciler) GetPublisher() (string, websocket.Publisher) {
-	return "gate.event", &socketPublisher{
+	return "gate", &socketPublisher{
 		gateQueue: s.gateQueue,
 	}
 }
diff --git a/pkg/controller/restore/reconciler.go b/pkg/controller/restore/reconciler.go
@@ -94,7 +94,7 @@ func (s *RestoreReconciler) GetPollInterval() func() time.Duration {
 }
 
 func (s *RestoreReconciler) GetPublisher() (string, websocket.Publisher) {
-	return "restore.event", &socketPublisher{
+	return "restore", &socketPublisher{
 		restoreQueue: s.restoreQueue,
 		restoreCache: s.restoreCache,
 	}
diff --git a/pkg/controller/sentinel/reconciler.go b/pkg/controller/sentinel/reconciler.go
@@ -88,7 +88,7 @@ func (r *SentinelReconciler) GetPollInterval() func() time.Duration {
 }
 
 func (r *SentinelReconciler) GetPublisher() (string, websocket.Publisher) {
-	return "sentinel.run.event", &socketPublisher{
+	return "sentinel_run", &socketPublisher{
 		sentinelRunQueue: r.sentinelQueue,
 		sentinelRunCache: r.sentinelCache,
 	}
diff --git a/pkg/controller/service/reconciler.go b/pkg/controller/service/reconciler.go
@@ -171,7 +171,7 @@ func (s *ServiceReconciler) GetPollInterval() func() time.Duration {
 }
 
 func (s *ServiceReconciler) GetPublisher() (string, websocket.Publisher) {
-	return "service.event", &socketPublisher{
+	return "service", &socketPublisher{
 		svcQueueGetter: func() workqueue.TypedRateLimitingInterface[string] { return s.svcQueue },
 		svcCache:       s.svcCache,
 		manCache:       s.manifestCache,
diff --git a/pkg/controller/stacks/reconciler.go b/pkg/controller/stacks/reconciler.go
@@ -88,7 +88,7 @@ func (r *StackReconciler) GetPollInterval() func() time.Duration {
 }
 
 func (r *StackReconciler) GetPublisher() (string, websocket.Publisher) {
-	return "stack.run.event", &socketPublisher{
+	return "stack_run", &socketPublisher{
 		stackRunQueue: r.stackQueue,
 		stackRunCache: r.stackCache,
 	}
diff --git a/pkg/controller/test/common.go b/pkg/controller/test/common.go
diff --git a/pkg/controller/v1/types.go b/pkg/controller/v1/types.go
diff --git a/pkg/websocket/socket.go b/pkg/websocket/socket.go
diff --git a/pkg/websocket/socket_test.go b/pkg/websocket/socket_test.go

Original file line number	Diff line number	Diff line change
`@@ -317,7 +317,7 @@ func registerKubeReconcilersOrDie(`
`317`	`317`	`if err := agentRuntimeReconciler.SetupWithManager(manager); err != nil {`
`318`	`318`	`setupLog.Error(err, "unable to create controller", "controller", "AgentRuntime")`
`319`	`319`	`}`
`320`		`- consoleManager.Socket.AddPublisher("agent.run.event", agentRuntimeReconciler)`
	`320`	`+ consoleManager.Socket.AddPublisher("agent_run", agentRuntimeReconciler)`
`321`	`321`
`322`	`322`	`if err := (&controller.AgentRunReconciler{`
`323`	`323`	`Client: manager.GetClient(),`