[client] Move metrics saver into its own metricsstages package

Relocate MetricsStages (per-connection metric stage timestamps) out of
the peer package into client/internal/peer/metricsstages.

.github/workflows/golang-test-linux.yml

@@ -166,7 +166,7 @@ jobs:
         # resolve; the grep then drops the broken package by path. Without -e,
         # go list aborts with empty stdout and `go test` falls back to the repo
         # root, which has no Go files.
-        run: CGO_ENABLED=1 GOARCH=${{ matrix.arch }} CI=true go test -coverprofile=coverage.txt -tags devcert -timeout 10m -p 1 $(go list -e ./... | grep -v -e /management -e /signal -e /relay -e /proxy -e /combined -e /client/ui)
+        run: CGO_ENABLED=1 GOARCH=${{ matrix.arch }} CI=true go test -coverprofile=coverage.txt -tags 'devcert privileged' -exec 'sudo --preserve-env=CI,CGO_ENABLED' -timeout 10m -p 1 $(go list -e ./... | grep -v -e /management -e /signal -e /relay -e /proxy -e /combined -e /client/ui -e /client/testutil/privileged)
 
       - name: Upload coverage reports to Codecov
         if: matrix.arch == 'amd64'

.github/workflows/release.yml

@@ -293,8 +293,11 @@ jobs:
           ${{ steps.goreleaser.outputs.artifacts }}
           JSON
 
+          # dockers_v2 artifacts have no top-level goarch field, so match the
+          # per-platform -amd64 tag suffix instead; it works for both the old
+          # dockers and the new dockers_v2 image naming.
           mapfile -t src_images < <(
-            jq -r '.[] | select(.type == "Docker Image") | select(.goarch == "amd64") | .name | select(startswith("ghcr.io/"))' /tmp/goreleaser-artifacts.json
+            jq -r '.[] | select(.type == "Docker Image") | .name | select(startswith("ghcr.io/") and endswith("-amd64"))' /tmp/goreleaser-artifacts.json
           )
 
           for src in "${src_images[@]}"; do

.gitignore

@@ -1,3 +1,4 @@
+.claude
 .idea
 .run
 *.iml

agent-network/README.md

@@ -1,16 +1,47 @@
 # NetBird Agent Network
 
-Agent Network is NetBird's access control layer for AI agents and the people who run
-them. It gives every agent a real identity, tied to your identity provider (IdP), and
-governs what it can reach — the LLM APIs and AI gateways it can call, and the internal
-resources it can access. Traffic flows only over the encrypted NetBird tunnel, scoped by
-policy, with no API keys to leak.
+Agent Network is NetBird's access control layer for AI agents and the people who run them. 
+It gives every agent a real identity, tied to an identity provider (IdP), and governs what it can reach: LLM APIs and 
+AI gateways it can call, and the internal resources it can access. Traffic flows only over the encrypted NetBird tunnel, 
+scoped by policy, with no API keys or other credentials to leak. It also gives you control over cost and token usage.
 
-> **Beta.** Agent Network is open source and can be self-hosted on your own
-> infrastructure.
+Because every LLM request passes through an
+identity-aware proxy, you can:
+
+- **Set spending and rate limits** per agent, per user, or per team — with hard caps
+  that stop requests once a budget is reached.
+- **Restrict models and providers** so agents can only call approved (and cost-appropriate)
+  endpoints, keeping expensive models off-limits unless explicitly allowed.
+- **Attribute usage** by tracking token consumption and cost per identity, group, or cost center so every
+  request is tied back to the agent and person responsible.
+- **Reuse your existing AI gateway** — point the proxy at a gateway you already run,
+  keeping its routing and config in place while it adds identity on top, so you skip
+  API key distribution.
+
+https://github.com/user-attachments/assets/44d18286-d8ab-49f8-a457-98ccd66f3268
+
+> **Beta.** Agent Network is in beta, but it's stable and already running in
+> production environments. It's fully open source and can be self-hosted on your own
+> infrastructure, with no vendor lock-in and no data leaving your environment.
 
 ## How it works
 
+Say you have a simple use case: your Engineering or IT team needs access to Claude Code or Codex, and you want visibility into usage plus the ability to enforce budgets.
+How can you do that without creating a dedicated API key for every team?
+
+With Agent Network you get a private endpoint inside your network, for example: https://mirror.netbird.ai
+Teams configure their agents to point to that endpoint instead of using individual API keys directly. 
+
+This endpoint is only reachable when users are connected to your NetBird network and authenticated through your IdP. Otherwise, it is not accessible from the public internet.
+You can then use this private endpoint to configure your AI agents, whether that is Claude Code, Codex, or another tool.
+
+## Quickstart
+
+Full step-by-step setup:
+**https://docs.netbird.io/agent-network/quickstart**
+
+## Architecture 
+
 Agent Network is built on two existing NetBird capabilities:
 
 - **Overlay network** — the encrypted WireGuard mesh between peers.
@@ -22,6 +53,9 @@ LLM traffic is routed through the proxy's identity-aware pipeline, while interna
 resources (databases, internal APIs, self-hosted models) are reached directly over
 peer-to-peer WireGuard tunnels, governed by the same identities and access policies.
 
+<img width="4720" height="2218" alt="image" src="https://github.com/user-attachments/assets/1afa5da1-4b82-4f8a-a7a8-f417efadf1eb" />
+
+
 ## Where the code lives
 
 There is no separate "agent-network" service — it reuses the reverse-proxy and management

client/android/env_list.go

@@ -10,7 +10,7 @@ var (
 	EnvKeyNBForceRelay = peer.EnvKeyNBForceRelay
 
 	// EnvKeyNBLazyConn Exported for Android java client to configure lazy connection
-	EnvKeyNBLazyConn = lazyconn.EnvEnableLazyConn
+	EnvKeyNBLazyConn = lazyconn.EnvLazyConn
 
 	// EnvKeyNBInactivityThreshold Exported for Android java client to configure connection inactivity threshold
 	EnvKeyNBInactivityThreshold = lazyconn.EnvInactivityThreshold

client/cmd/root.go

@@ -71,12 +71,14 @@ var (
 	extraIFaceBlackList     []string
 	anonymizeFlag           bool
 	dnsRouteInterval        time.Duration
-	lazyConnEnabled         bool
-	mtu                     uint16
-	profilesDisabled        bool
-	updateSettingsDisabled  bool
-	captureEnabled          bool
-	networksDisabled        bool
+	// lazyConnEnabled is the parse target for the deprecated --enable-lazy-connection
+	// flag. The flag is inert; the value is no longer read (use NB_LAZY_CONN instead).
+	lazyConnEnabled        bool
+	mtu                    uint16
+	profilesDisabled       bool
+	updateSettingsDisabled bool
+	captureEnabled         bool
+	networksDisabled       bool
 
 	rootCmd = &cobra.Command{
 		Use:          "netbird",
@@ -210,7 +212,8 @@ func init() {
 	upCmd.PersistentFlags().BoolVar(&rosenpassEnabled, enableRosenpassFlag, false, "[Experimental] Enable Rosenpass feature. If enabled, the connection will be post-quantum secured via Rosenpass.")
 	upCmd.PersistentFlags().BoolVar(&rosenpassPermissive, rosenpassPermissiveFlag, false, "[Experimental] Enable Rosenpass in permissive mode to allow this peer to accept WireGuard connections without requiring Rosenpass functionality from peers that do not have Rosenpass enabled.")
 	upCmd.PersistentFlags().BoolVar(&autoConnectDisabled, disableAutoConnectFlag, false, "Disables auto-connect feature. If enabled, then the client won't connect automatically when the service starts.")
-	upCmd.PersistentFlags().BoolVar(&lazyConnEnabled, enableLazyConnectionFlag, false, "[Experimental] Enable the lazy connection feature. If enabled, the client will establish connections on-demand. Note: this setting may be overridden by management configuration.")
+	upCmd.PersistentFlags().BoolVar(&lazyConnEnabled, enableLazyConnectionFlag, false, "Deprecated: no longer used. Lazy connections are controlled by the server and the NB_LAZY_CONN environment variable.")
+	_ = upCmd.PersistentFlags().MarkDeprecated(enableLazyConnectionFlag, "no longer used; lazy connections are controlled by the server and the NB_LAZY_CONN environment variable")
 
 }
 

client/cmd/service.go

@@ -5,6 +5,7 @@ package cmd
 import (
 	"context"
 	"fmt"
+	"net/http"
 	"runtime"
 	"strings"
 	"sync"
@@ -22,15 +23,21 @@ var serviceCmd = &cobra.Command{
 	Short: "Manage the NetBird daemon service",
 }
 
+const defaultJSONSocket = "unix:///var/run/netbird-http.sock"
+
 var (
-	serviceName    string
-	serviceEnvVars []string
+	serviceName      string
+	serviceEnvVars   []string
+	jsonSocket       string
+	enableJSONSocket bool
 )
 
 type program struct {
 	ctx              context.Context
 	cancel           context.CancelFunc
 	serv             *grpc.Server
+	jsonServ         *http.Server
+	jsonServMu       sync.Mutex
 	serverInstance   *server.Server
 	serverInstanceMu sync.Mutex
 }
@@ -46,6 +53,8 @@ func init() {
 	serviceCmd.PersistentFlags().BoolVar(&updateSettingsDisabled, "disable-update-settings", false, "Disables update settings feature. If enabled, the client will not be able to change or edit any settings. To persist this setting, use: netbird service install --disable-update-settings")
 	serviceCmd.PersistentFlags().BoolVar(&captureEnabled, "enable-capture", false, "Enables packet capture via 'netbird debug capture'. To persist, use: netbird service install --enable-capture")
 	serviceCmd.PersistentFlags().BoolVar(&networksDisabled, "disable-networks", false, "Disables network selection. If enabled, the client will not allow listing, selecting, or deselecting networks. To persist, use: netbird service install --disable-networks")
+	serviceCmd.PersistentFlags().BoolVar(&enableJSONSocket, "enable-json-socket", false, "Enables the HTTP/JSON API socket served by grpc-gateway. To persist, use: netbird service install --enable-json-socket")
+	serviceCmd.PersistentFlags().StringVar(&jsonSocket, "json-socket", defaultJSONSocket, "HTTP/JSON API socket address [unix|tcp]://[path|host:port]. Requires --enable-json-socket to serve. To persist, use: netbird service install --enable-json-socket --json-socket")
 
 	rootCmd.PersistentFlags().StringVarP(&serviceName, "service", "s", defaultServiceName, "Netbird system service name")
 	serviceEnvDesc := `Sets extra environment variables for the service. ` +

client/cmd/service_controller.go

@@ -5,9 +5,6 @@ package cmd
 import (
 	"context"
 	"fmt"
-	"net"
-	"os"
-	"strings"
 	"time"
 
 	"github.com/kardianos/service"
@@ -22,41 +19,56 @@ import (
 	"github.com/netbirdio/netbird/util"
 )
 
+func validateJSONSocketFlags() error {
+	if serviceCmd.PersistentFlags().Changed("json-socket") && !enableJSONSocket {
+		return fmt.Errorf("--json-socket requires --enable-json-socket to configure the daemon JSON gateway")
+	}
+	return nil
+}
+
 func (p *program) Start(svc service.Service) error {
 	// Start should not block. Do the actual work async.
 	log.Info("starting NetBird service") //nolint
 
+	if err := validateJSONSocketFlags(); err != nil {
+		return err
+	}
+
 	// Collect static system and platform information
 	system.UpdateStaticInfoAsync()
 
 	// in any case, even if configuration does not exists we run daemon to serve CLI gRPC API.
 	p.serv = grpc.NewServer()
 
-	split := strings.Split(daemonAddr, "://")
-	switch split[0] {
-	case "unix":
-		// cleanup failed close
-		stat, err := os.Stat(split[1])
-		if err == nil && !stat.IsDir() {
-			if err := os.Remove(split[1]); err != nil {
-				log.Debugf("remove socket file: %v", err)
-			}
-		}
-	case "tcp":
-	default:
-		return fmt.Errorf("unsupported daemon address protocol: %v", split[0])
-	}
-
-	listen, err := net.Listen(split[0], split[1])
+	daemonListener, err := listenOnAddress(daemonAddr)
 	if err != nil {
 		return fmt.Errorf("listen daemon interface: %w", err)
 	}
-	go func() {
-		defer listen.Close()
 
-		if split[0] == "unix" {
-			if err := os.Chmod(split[1], 0666); err != nil {
-				log.Errorf("failed setting daemon permissions: %v", split[1])
+	var jsonListener *socketListener
+	if enableJSONSocket {
+		jsonListener, err = listenOnAddress(jsonSocket)
+		if err != nil {
+			_ = daemonListener.Close()
+			return fmt.Errorf("listen daemon JSON interface: %w", err)
+		}
+	} else {
+		removeStaleUnixSocketForAddress(jsonSocket)
+	}
+
+	go func() {
+		defer daemonListener.Close()
+		if jsonListener != nil {
+			defer jsonListener.Close()
+		}
+
+		if err := daemonListener.chmodUnixSocket("daemon"); err != nil {
+			log.Error(err)
+			return
+		}
+		if jsonListener != nil {
+			if err := jsonListener.chmodUnixSocket("daemon JSON"); err != nil {
+				log.Error(err)
 				return
 			}
 		}
@@ -71,8 +83,16 @@ func (p *program) Start(svc service.Service) error {
 		p.serverInstance = serverInstance
 		p.serverInstanceMu.Unlock()
 
-		log.Printf("started daemon server: %v", split[1])
-		if err := p.serv.Serve(listen); err != nil {
+		if jsonListener != nil {
+			if err := p.startJSONGateway(jsonListener, daemonAddr); err != nil {
+				log.Fatalf("failed to start daemon JSON server: %v", err)
+			}
+		} else {
+			log.Debug("daemon JSON socket disabled")
+		}
+
+		log.Printf("started daemon server: %v", daemonListener.address)
+		if err := p.serv.Serve(daemonListener.Listener); err != nil {
 			log.Errorf("failed to serve daemon requests: %v", err)
 		}
 	}()
@@ -92,6 +112,20 @@ func (p *program) Stop(srv service.Service) error {
 
 	p.cancel()
 
+	p.jsonServMu.Lock()
+	jsonServ := p.jsonServ
+	p.jsonServMu.Unlock()
+	if jsonServ != nil {
+		shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 2*time.Second)
+		if err := jsonServ.Shutdown(shutdownCtx); err != nil {
+			log.Errorf("failed to stop daemon JSON server gracefully: %v", err)
+			if err := jsonServ.Close(); err != nil {
+				log.Errorf("failed to close daemon JSON server: %v", err)
+			}
+		}
+		shutdownCancel()
+	}
+
 	if p.serv != nil {
 		p.serv.Stop()
 	}
@@ -148,6 +182,9 @@ var runCmd = &cobra.Command{
 		if err != nil {
 			return err
 		}
+		if err := validateJSONSocketFlags(); err != nil {
+			return err
+		}
 
 		return s.Run()
 	},
@@ -162,6 +199,9 @@ var startCmd = &cobra.Command{
 		if err != nil {
 			return err
 		}
+		if err := validateJSONSocketFlags(); err != nil {
+			return err
+		}
 
 		if err := s.Start(); err != nil {
 			return fmt.Errorf("start service: %w", err)
@@ -198,6 +238,9 @@ var restartCmd = &cobra.Command{
 		if err != nil {
 			return err
 		}
+		if err := validateJSONSocketFlags(); err != nil {
+			return err
+		}
 
 		if err := s.Restart(); err != nil {
 			return fmt.Errorf("restart service: %w", err)

client/cmd/service_installer.go

@@ -67,6 +67,10 @@ func buildServiceArguments() []string {
 		args = append(args, "--disable-networks")
 	}
 
+	if enableJSONSocket {
+		args = append(args, "--enable-json-socket", "--json-socket", jsonSocket)
+	}
+
 	return args
 }
 
@@ -106,6 +110,10 @@ func configurePlatformSpecificSettings(svcConfig *service.Config) error {
 
 // Create fully configured service config for install/reconfigure
 func createServiceConfigForInstall() (*service.Config, error) {
+	if err := validateJSONSocketFlags(); err != nil {
+		return nil, err
+	}
+
 	svcConfig, err := newSVCConfig()
 	if err != nil {
 		return nil, fmt.Errorf("create service config: %w", err)

client/cmd/service_json_gateway.go

@@ -0,0 +1,52 @@
+//go:build !ios && !android
+
+package cmd
+
+import (
+	"context"
+	"errors"
+	"net"
+	"net/http"
+	"strings"
+	"time"
+
+	"github.com/grpc-ecosystem/grpc-gateway/v2/runtime"
+	log "github.com/sirupsen/logrus"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+
+	"github.com/netbirdio/netbird/client/proto"
+)
+
+func grpcGatewayEndpoint(addr string) string {
+	return strings.TrimPrefix(addr, "tcp://")
+}
+
+func (p *program) startJSONGateway(jsonListener *socketListener, daemonEndpoint string) error {
+	mux := runtime.NewServeMux()
+	opts := []grpc.DialOption{grpc.WithTransportCredentials(insecure.NewCredentials())}
+	if err := proto.RegisterDaemonServiceHandlerFromEndpoint(p.ctx, mux, grpcGatewayEndpoint(daemonEndpoint), opts); err != nil {
+		return err
+	}
+
+	jsonServer := &http.Server{
+		Handler:           mux,
+		ReadHeaderTimeout: 5 * time.Second,
+		BaseContext: func(net.Listener) context.Context {
+			return p.ctx
+		},
+	}
+
+	p.jsonServMu.Lock()
+	p.jsonServ = jsonServer
+	p.jsonServMu.Unlock()
+
+	go func() {
+		log.Printf("started daemon JSON server: %v", jsonListener.address)
+		if err := jsonServer.Serve(jsonListener.Listener); err != nil && !errors.Is(err, http.ErrServerClosed) {
+			log.Errorf("failed to serve daemon JSON requests: %v", err)
+		}
+	}()
+
+	return nil
+}

client/cmd/service_json_socket_test.go

@@ -0,0 +1,176 @@
+//go:build !ios && !android
+
+package cmd
+
+import (
+	"net"
+	"os"
+	"path/filepath"
+	"runtime"
+	"testing"
+
+	"github.com/spf13/cobra"
+	"github.com/spf13/pflag"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func preserveJSONSocketTestState(t *testing.T) {
+	t.Helper()
+
+	origJSONSocket := jsonSocket
+	origEnableJSONSocket := enableJSONSocket
+	origChanged := map[string]bool{}
+	serviceCmd.PersistentFlags().VisitAll(func(flag *pflag.Flag) {
+		origChanged[flag.Name] = flag.Changed
+	})
+
+	t.Cleanup(func() {
+		jsonSocket = origJSONSocket
+		enableJSONSocket = origEnableJSONSocket
+		serviceCmd.PersistentFlags().VisitAll(func(flag *pflag.Flag) {
+			flag.Changed = origChanged[flag.Name]
+		})
+	})
+}
+
+func TestJSONSocketFlagsArePositiveEnableOnly(t *testing.T) {
+	assert.NotNil(t, serviceCmd.PersistentFlags().Lookup("enable-json-socket"))
+	assert.NotNil(t, serviceCmd.PersistentFlags().Lookup("json-socket"))
+	assert.Nil(t, serviceCmd.PersistentFlags().Lookup("disable-json-socket"))
+	assert.Equal(t, "false", serviceCmd.PersistentFlags().Lookup("enable-json-socket").DefValue)
+}
+
+func TestBuildServiceArgumentsDefaultDisablesJSONSocket(t *testing.T) {
+	preserveJSONSocketTestState(t)
+
+	enableJSONSocket = false
+	jsonSocket = "tcp://127.0.0.1:8080"
+
+	args := buildServiceArguments()
+
+	assert.NotContains(t, args, "--enable-json-socket")
+	assert.NotContains(t, args, "--json-socket")
+}
+
+func TestBuildServiceArgumentsIncludesJSONSocketWhenEnabled(t *testing.T) {
+	preserveJSONSocketTestState(t)
+
+	enableJSONSocket = true
+	jsonSocket = "tcp://127.0.0.1:8080"
+
+	args := buildServiceArguments()
+
+	enableIndex := indexOfArg(args, "--enable-json-socket")
+	jsonIndex := indexOfArg(args, "--json-socket")
+	require.NotEqual(t, -1, enableIndex)
+	require.NotEqual(t, -1, jsonIndex)
+	require.Less(t, enableIndex, jsonIndex)
+	require.Less(t, jsonIndex+1, len(args))
+	assert.Equal(t, "tcp://127.0.0.1:8080", args[jsonIndex+1])
+}
+
+func TestJSONSocketWithoutEnableValidation(t *testing.T) {
+	preserveJSONSocketTestState(t)
+
+	enableJSONSocket = false
+	require.NoError(t, serviceCmd.PersistentFlags().Set("json-socket", "tcp://127.0.0.1:8080"))
+
+	err := validateJSONSocketFlags()
+
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "--enable-json-socket")
+}
+
+func TestJSONSocketWithEnableValidation(t *testing.T) {
+	preserveJSONSocketTestState(t)
+
+	require.NoError(t, serviceCmd.PersistentFlags().Set("enable-json-socket", "true"))
+	require.NoError(t, serviceCmd.PersistentFlags().Set("json-socket", "tcp://127.0.0.1:8080"))
+
+	assert.NoError(t, validateJSONSocketFlags())
+}
+
+func TestJSONSocketServiceParamsPersistEnableAndAddress(t *testing.T) {
+	preserveJSONSocketTestState(t)
+	serviceCmd.PersistentFlags().VisitAll(func(flag *pflag.Flag) {
+		flag.Changed = false
+	})
+
+	enableJSONSocket = true
+	jsonSocket = "tcp://127.0.0.1:8080"
+
+	params := currentServiceParams()
+	require.True(t, params.EnableJSONSocket)
+	require.Equal(t, "tcp://127.0.0.1:8080", params.JSONSocket)
+
+	enableJSONSocket = false
+	jsonSocket = defaultJSONSocket
+	applyServiceParams(testServiceEnvCommand(), params)
+
+	assert.True(t, enableJSONSocket)
+	assert.Equal(t, "tcp://127.0.0.1:8080", jsonSocket)
+}
+
+func TestRemoveStaleUnixSocketDoesNotRemoveRegularFile(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "netbird-http.sock")
+	require.NoError(t, os.WriteFile(path, []byte("not a socket"), 0600))
+
+	removeStaleUnixSocket(path)
+
+	data, err := os.ReadFile(path)
+	require.NoError(t, err)
+	assert.Equal(t, []byte("not a socket"), data)
+}
+
+func TestRemoveStaleUnixSocketRemovesSocket(t *testing.T) {
+	if runtime.GOOS == "windows" {
+		t.Skip("unix sockets are not available on Windows")
+	}
+
+	path := filepath.Join(t.TempDir(), "netbird-http.sock")
+	addr := &net.UnixAddr{Name: path, Net: "unix"}
+	listener, err := net.ListenUnix("unix", addr)
+	require.NoError(t, err)
+	listener.SetUnlinkOnClose(false)
+	require.NoError(t, listener.Close())
+
+	_, err = os.Lstat(path)
+	require.NoError(t, err, "test setup must leave a stale Unix socket path")
+
+	removeStaleUnixSocket(path)
+
+	_, err = os.Lstat(path)
+	assert.True(t, os.IsNotExist(err), "expected stale Unix socket to be removed, got %v", err)
+}
+
+func TestRemoveStaleUnixSocketDoesNotRemoveLiveSocket(t *testing.T) {
+	if runtime.GOOS == "windows" {
+		t.Skip("unix sockets are not available on Windows")
+	}
+
+	path := filepath.Join(t.TempDir(), "netbird-http.sock")
+	listener, err := net.Listen("unix", path)
+	require.NoError(t, err)
+	defer listener.Close()
+
+	removeStaleUnixSocket(path)
+
+	_, err = os.Lstat(path)
+	assert.NoError(t, err, "expected live Unix socket to be preserved")
+}
+
+func testServiceEnvCommand() *cobra.Command {
+	cmd := &cobra.Command{}
+	cmd.Flags().StringSlice("service-env", nil, "")
+	return cmd
+}
+
+func indexOfArg(args []string, arg string) int {
+	for i, candidate := range args {
+		if candidate == arg {
+			return i
+		}
+	}
+	return -1
+}

client/cmd/service_params.go

@@ -23,6 +23,7 @@ const serviceParamsFile = "service.json"
 type serviceParams struct {
 	LogLevel              string            `json:"log_level"`
 	DaemonAddr            string            `json:"daemon_addr"`
+	JSONSocket            string            `json:"json_socket"`
 	ManagementURL         string            `json:"management_url,omitempty"`
 	ConfigPath            string            `json:"config_path,omitempty"`
 	LogFiles              []string          `json:"log_files,omitempty"`
@@ -30,6 +31,7 @@ type serviceParams struct {
 	DisableUpdateSettings bool              `json:"disable_update_settings,omitempty"`
 	EnableCapture         bool              `json:"enable_capture,omitempty"`
 	DisableNetworks       bool              `json:"disable_networks,omitempty"`
+	EnableJSONSocket      bool              `json:"enable_json_socket,omitempty"`
 	ServiceEnvVars        map[string]string `json:"service_env_vars,omitempty"`
 }
 
@@ -75,6 +77,7 @@ func currentServiceParams() *serviceParams {
 	params := &serviceParams{
 		LogLevel:              logLevel,
 		DaemonAddr:            daemonAddr,
+		JSONSocket:            jsonSocket,
 		ManagementURL:         managementURL,
 		ConfigPath:            configPath,
 		LogFiles:              logFiles,
@@ -82,6 +85,7 @@ func currentServiceParams() *serviceParams {
 		DisableUpdateSettings: updateSettingsDisabled,
 		EnableCapture:         captureEnabled,
 		DisableNetworks:       networksDisabled,
+		EnableJSONSocket:      enableJSONSocket,
 	}
 
 	if len(serviceEnvVars) > 0 {
@@ -113,9 +117,8 @@ func applyServiceParams(cmd *cobra.Command, params *serviceParams) {
 		return
 	}
 
-	// For fields with non-empty defaults (log-level, daemon-addr), keep the
-	// != "" guard so that an older service.json missing the field doesn't
-	// clobber the default with an empty string.
+	// For fields with non-empty defaults, keep the != "" guard so that an older
+	// service.json missing the field doesn't clobber the default with an empty string.
 	if !rootCmd.PersistentFlags().Changed("log-level") && params.LogLevel != "" {
 		logLevel = params.LogLevel
 	}
@@ -124,6 +127,14 @@ func applyServiceParams(cmd *cobra.Command, params *serviceParams) {
 		daemonAddr = params.DaemonAddr
 	}
 
+	if !serviceCmd.PersistentFlags().Changed("json-socket") && params.JSONSocket != "" {
+		jsonSocket = params.JSONSocket
+	}
+
+	if !serviceCmd.PersistentFlags().Changed("enable-json-socket") {
+		enableJSONSocket = params.EnableJSONSocket
+	}
+
 	// For optional fields where empty means "use default", always apply so
 	// that an explicit clear (--management-url "") persists across reinstalls.
 	if !rootCmd.PersistentFlags().Changed("management-url") {

client/cmd/service_params_test.go

@@ -41,6 +41,8 @@ func TestSaveAndLoadServiceParams(t *testing.T) {
 	params := &serviceParams{
 		LogLevel:              "debug",
 		DaemonAddr:            "unix:///var/run/netbird.sock",
+		JSONSocket:            "tcp://127.0.0.1:8080",
+		EnableJSONSocket:      true,
 		ManagementURL:         "https://my.server.com",
 		ConfigPath:            "/etc/netbird/config.json",
 		LogFiles:              []string{"/var/log/netbird/client.log", "console"},
@@ -63,6 +65,8 @@ func TestSaveAndLoadServiceParams(t *testing.T) {
 
 	assert.Equal(t, params.LogLevel, loaded.LogLevel)
 	assert.Equal(t, params.DaemonAddr, loaded.DaemonAddr)
+	assert.Equal(t, params.JSONSocket, loaded.JSONSocket)
+	assert.Equal(t, params.EnableJSONSocket, loaded.EnableJSONSocket)
 	assert.Equal(t, params.ManagementURL, loaded.ManagementURL)
 	assert.Equal(t, params.ConfigPath, loaded.ConfigPath)
 	assert.Equal(t, params.LogFiles, loaded.LogFiles)
@@ -101,6 +105,8 @@ func TestLoadServiceParams_InvalidJSON(t *testing.T) {
 func TestCurrentServiceParams(t *testing.T) {
 	origLogLevel := logLevel
 	origDaemonAddr := daemonAddr
+	origJSONSocket := jsonSocket
+	origEnableJSONSocket := enableJSONSocket
 	origManagementURL := managementURL
 	origConfigPath := configPath
 	origLogFiles := logFiles
@@ -110,6 +116,8 @@ func TestCurrentServiceParams(t *testing.T) {
 	t.Cleanup(func() {
 		logLevel = origLogLevel
 		daemonAddr = origDaemonAddr
+		jsonSocket = origJSONSocket
+		enableJSONSocket = origEnableJSONSocket
 		managementURL = origManagementURL
 		configPath = origConfigPath
 		logFiles = origLogFiles
@@ -120,6 +128,8 @@ func TestCurrentServiceParams(t *testing.T) {
 
 	logLevel = "trace"
 	daemonAddr = "tcp://127.0.0.1:9999"
+	jsonSocket = "tcp://127.0.0.1:8080"
+	enableJSONSocket = true
 	managementURL = "https://mgmt.example.com"
 	configPath = "/tmp/test-config.json"
 	logFiles = []string{"/tmp/test.log"}
@@ -131,6 +141,8 @@ func TestCurrentServiceParams(t *testing.T) {
 
 	assert.Equal(t, "trace", params.LogLevel)
 	assert.Equal(t, "tcp://127.0.0.1:9999", params.DaemonAddr)
+	assert.Equal(t, "tcp://127.0.0.1:8080", params.JSONSocket)
+	assert.True(t, params.EnableJSONSocket)
 	assert.Equal(t, "https://mgmt.example.com", params.ManagementURL)
 	assert.Equal(t, "/tmp/test-config.json", params.ConfigPath)
 	assert.Equal(t, []string{"/tmp/test.log"}, params.LogFiles)
@@ -142,6 +154,8 @@ func TestCurrentServiceParams(t *testing.T) {
 func TestApplyServiceParams_OnlyUnchangedFlags(t *testing.T) {
 	origLogLevel := logLevel
 	origDaemonAddr := daemonAddr
+	origJSONSocket := jsonSocket
+	origEnableJSONSocket := enableJSONSocket
 	origManagementURL := managementURL
 	origConfigPath := configPath
 	origLogFiles := logFiles
@@ -151,6 +165,8 @@ func TestApplyServiceParams_OnlyUnchangedFlags(t *testing.T) {
 	t.Cleanup(func() {
 		logLevel = origLogLevel
 		daemonAddr = origDaemonAddr
+		jsonSocket = origJSONSocket
+		enableJSONSocket = origEnableJSONSocket
 		managementURL = origManagementURL
 		configPath = origConfigPath
 		logFiles = origLogFiles
@@ -162,6 +178,8 @@ func TestApplyServiceParams_OnlyUnchangedFlags(t *testing.T) {
 	// Reset all flags to defaults.
 	logLevel = "info"
 	daemonAddr = "unix:///var/run/netbird.sock"
+	jsonSocket = defaultJSONSocket
+	enableJSONSocket = false
 	managementURL = ""
 	configPath = "/etc/netbird/config.json"
 	logFiles = []string{"/var/log/netbird/client.log"}
@@ -184,6 +202,8 @@ func TestApplyServiceParams_OnlyUnchangedFlags(t *testing.T) {
 	saved := &serviceParams{
 		LogLevel:              "debug",
 		DaemonAddr:            "tcp://127.0.0.1:5555",
+		JSONSocket:            "tcp://127.0.0.1:8080",
+		EnableJSONSocket:      true,
 		ManagementURL:         "https://saved.example.com",
 		ConfigPath:            "/saved/config.json",
 		LogFiles:              []string{"/saved/client.log"},
@@ -201,6 +221,8 @@ func TestApplyServiceParams_OnlyUnchangedFlags(t *testing.T) {
 
 	// All other fields were not Changed, so they should use saved values.
 	assert.Equal(t, "tcp://127.0.0.1:5555", daemonAddr)
+	assert.Equal(t, "tcp://127.0.0.1:8080", jsonSocket)
+	assert.True(t, enableJSONSocket)
 	assert.Equal(t, "https://saved.example.com", managementURL)
 	assert.Equal(t, "/saved/config.json", configPath)
 	assert.Equal(t, []string{"/saved/client.log"}, logFiles)
@@ -212,14 +234,17 @@ func TestApplyServiceParams_OnlyUnchangedFlags(t *testing.T) {
 func TestApplyServiceParams_BooleanRevertToFalse(t *testing.T) {
 	origProfilesDisabled := profilesDisabled
 	origUpdateSettingsDisabled := updateSettingsDisabled
+	origEnableJSONSocket := enableJSONSocket
 	t.Cleanup(func() {
 		profilesDisabled = origProfilesDisabled
 		updateSettingsDisabled = origUpdateSettingsDisabled
+		enableJSONSocket = origEnableJSONSocket
 	})
 
 	// Simulate current state where booleans are true (e.g. set by previous install).
 	profilesDisabled = true
 	updateSettingsDisabled = true
+	enableJSONSocket = true
 
 	// Reset Changed state so flags appear unset.
 	serviceCmd.PersistentFlags().VisitAll(func(f *pflag.Flag) {
@@ -238,6 +263,7 @@ func TestApplyServiceParams_BooleanRevertToFalse(t *testing.T) {
 
 	assert.False(t, profilesDisabled, "saved false should override current true")
 	assert.False(t, updateSettingsDisabled, "saved false should override current true")
+	assert.False(t, enableJSONSocket, "saved false should override current true")
 }
 
 func TestApplyServiceParams_ClearManagementURL(t *testing.T) {
@@ -530,6 +556,7 @@ func fieldToGlobalVar(field string) string {
 	m := map[string]string{
 		"LogLevel":              "logLevel",
 		"DaemonAddr":            "daemonAddr",
+		"JSONSocket":            "jsonSocket",
 		"ManagementURL":         "managementURL",
 		"ConfigPath":            "configPath",
 		"LogFiles":              "logFiles",
@@ -537,6 +564,7 @@ func fieldToGlobalVar(field string) string {
 		"DisableUpdateSettings": "updateSettingsDisabled",
 		"EnableCapture":         "captureEnabled",
 		"DisableNetworks":       "networksDisabled",
+		"EnableJSONSocket":      "enableJSONSocket",
 		"ServiceEnvVars":        "serviceEnvVars",
 	}
 	if v, ok := m[field]; ok {

client/cmd/service_socket.go

@@ -0,0 +1,111 @@
+//go:build !ios && !android
+
+package cmd
+
+import (
+	"errors"
+	"fmt"
+	"net"
+	"os"
+	"strings"
+	"syscall"
+	"time"
+
+	log "github.com/sirupsen/logrus"
+)
+
+type socketListener struct {
+	net.Listener
+	network string
+	address string
+}
+
+func listenOnAddress(addr string) (*socketListener, error) {
+	network, address, err := parseListenAddress(addr)
+	if err != nil {
+		return nil, err
+	}
+
+	if network == "unix" {
+		removeStaleUnixSocket(address)
+	}
+
+	listener, err := net.Listen(network, address)
+	if err != nil {
+		return nil, err
+	}
+
+	return &socketListener{Listener: listener, network: network, address: address}, nil
+}
+
+func parseListenAddress(addr string) (string, string, error) {
+	network, address, ok := strings.Cut(addr, "://")
+	if !ok || network == "" || address == "" {
+		return "", "", fmt.Errorf("address must be in [unix|tcp]://[path|host:port] format: %q", addr)
+	}
+
+	switch network {
+	case "unix", "tcp":
+		return network, address, nil
+	default:
+		return "", "", fmt.Errorf("unsupported daemon address protocol: %v", network)
+	}
+}
+
+func removeStaleUnixSocket(path string) {
+	stat, err := os.Lstat(path)
+	if err != nil {
+		if !os.IsNotExist(err) {
+			log.Debugf("stat socket file: %v", err)
+		}
+		return
+	}
+
+	if stat.Mode()&os.ModeSocket == 0 {
+		return
+	}
+
+	if !isStaleUnixSocket(path) {
+		return
+	}
+
+	if err := os.Remove(path); err != nil {
+		log.Debugf("remove socket file: %v", err)
+	}
+}
+
+func isStaleUnixSocket(path string) bool {
+	conn, err := net.DialTimeout("unix", path, 100*time.Millisecond)
+	if err == nil {
+		if closeErr := conn.Close(); closeErr != nil {
+			log.Debugf("close unix socket probe: %v", closeErr)
+		}
+		return false
+	}
+
+	if os.IsNotExist(err) || os.IsPermission(err) || os.IsTimeout(err) {
+		log.Debugf("not removing unix socket %s after probe error: %v", path, err)
+		return false
+	}
+
+	return errors.Is(err, syscall.ECONNREFUSED)
+}
+
+func removeStaleUnixSocketForAddress(addr string) {
+	network, address, err := parseListenAddress(addr)
+	if err != nil || network != "unix" {
+		return
+	}
+	removeStaleUnixSocket(address)
+}
+
+func (l *socketListener) chmodUnixSocket(description string) error {
+	if l == nil || l.network != "unix" {
+		return nil
+	}
+
+	if err := os.Chmod(l.address, 0666); err != nil {
+		return fmt.Errorf("failed setting %s permissions for %s: %w", description, l.address, err)
+	}
+	return nil
+}

client/cmd/up.go

@@ -479,10 +479,6 @@ func setupSetConfigReq(customDNSAddressConverted []byte, cmd *cobra.Command, pro
 		req.DisableIpv6 = &disableIPv6
 	}
 
-	if cmd.Flag(enableLazyConnectionFlag).Changed {
-		req.LazyConnectionEnabled = &lazyConnEnabled
-	}
-
 	return &req
 }
 
@@ -600,9 +596,6 @@ func setupConfig(customDNSAddressConverted []byte, cmd *cobra.Command, configFil
 		ic.DisableIPv6 = &disableIPv6
 	}
 
-	if cmd.Flag(enableLazyConnectionFlag).Changed {
-		ic.LazyConnectionEnabled = &lazyConnEnabled
-	}
 	return &ic, nil
 }
 
@@ -718,9 +711,6 @@ func setupLoginRequest(providedSetupKey string, customDNSAddressConverted []byte
 		loginRequest.DisableIpv6 = &disableIPv6
 	}
 
-	if cmd.Flag(enableLazyConnectionFlag).Changed {
-		loginRequest.LazyConnectionEnabled = &lazyConnEnabled
-	}
 	return &loginRequest, nil
 }
 

client/iface/configurer/kernel_unix.go

@@ -17,12 +17,15 @@ import (
 
 type KernelConfigurer struct {
 	deviceName string
+	statsCache *statsCache
 }
 
 func NewKernelConfigurer(deviceName string) *KernelConfigurer {
-	return &KernelConfigurer{
+	c := &KernelConfigurer{
 		deviceName: deviceName,
 	}
+	c.statsCache = newStatsCache(statsCacheTTL, c.fetchStats)
+	return c
 }
 
 func (c *KernelConfigurer) ConfigureInterface(privateKey string, port int) error {
@@ -246,12 +249,6 @@ func (c *KernelConfigurer) configure(config wgtypes.Config) error {
 		}
 	}()
 
-	// validate if device with name exists
-	_, err = wg.Device(c.deviceName)
-	if err != nil {
-		return err
-	}
-
 	return wg.ConfigureDevice(c.deviceName, config)
 }
 
@@ -300,6 +297,14 @@ func (c *KernelConfigurer) FullStats() (*Stats, error) {
 }
 
 func (c *KernelConfigurer) GetStats() (map[string]WGStats, error) {
+	return c.statsCache.get()
+}
+
+func (c *KernelConfigurer) LastActivities() map[string]monotime.Time {
+	return nil
+}
+
+func (c *KernelConfigurer) fetchStats() (map[string]WGStats, error) {
 	stats := make(map[string]WGStats)
 	wg, err := wgctrl.New()
 	if err != nil {
@@ -326,7 +331,3 @@ func (c *KernelConfigurer) GetStats() (map[string]WGStats, error) {
 	}
 	return stats, nil
 }
-
-func (c *KernelConfigurer) LastActivities() map[string]monotime.Time {
-	return nil
-}

client/iface/configurer/stats_cache.go

@@ -0,0 +1,52 @@
+package configurer
+
+import (
+	"sync"
+	"time"
+
+	"golang.org/x/sync/singleflight"
+)
+
+const statsCacheTTL = 1 * time.Second
+
+type statsCache struct {
+	ttl   time.Duration
+	fetch func() (map[string]WGStats, error)
+
+	mu       sync.RWMutex
+	value    map[string]WGStats
+	expireAt time.Time
+
+	sf singleflight.Group
+}
+
+func newStatsCache(ttl time.Duration, fetch func() (map[string]WGStats, error)) *statsCache {
+	return &statsCache{ttl: ttl, fetch: fetch}
+}
+
+func (c *statsCache) get() (map[string]WGStats, error) {
+	c.mu.RLock()
+	if c.value != nil && time.Now().Before(c.expireAt) {
+		value := c.value
+		c.mu.RUnlock()
+		return value, nil
+	}
+	c.mu.RUnlock()
+
+	value, err, _ := c.sf.Do("stats", func() (interface{}, error) {
+		res, err := c.fetch()
+		if err != nil {
+			return nil, err
+		}
+
+		c.mu.Lock()
+		c.value = res
+		c.expireAt = time.Now().Add(c.ttl)
+		c.mu.Unlock()
+		return res, nil
+	})
+	if err != nil {
+		return nil, err
+	}
+	return value.(map[string]WGStats), nil
+}

client/iface/configurer/stats_cache_test.go

@@ -0,0 +1,70 @@
+package configurer
+
+import (
+	"errors"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestStatsCache_CachesWithinTTL(t *testing.T) {
+	var calls atomic.Int64
+	c := newStatsCache(50*time.Millisecond, func() (map[string]WGStats, error) {
+		calls.Add(1)
+		return map[string]WGStats{"p": {}}, nil
+	})
+
+	for i := 0; i < 10; i++ {
+		_, err := c.get()
+		require.NoError(t, err)
+	}
+	require.Equal(t, int64(1), calls.Load(), "within TTL only one underlying fetch")
+
+	time.Sleep(60 * time.Millisecond)
+	_, err := c.get()
+	require.NoError(t, err)
+	require.Equal(t, int64(2), calls.Load(), "after TTL expiry a fresh fetch happens")
+}
+
+func TestStatsCache_SingleFlight(t *testing.T) {
+	var calls atomic.Int64
+	release := make(chan struct{})
+	c := newStatsCache(time.Minute, func() (map[string]WGStats, error) {
+		calls.Add(1)
+		<-release
+		return map[string]WGStats{}, nil
+	})
+
+	const n = 50
+	var wg sync.WaitGroup
+	wg.Add(n)
+	for i := 0; i < n; i++ {
+		go func() {
+			defer wg.Done()
+			_, _ = c.get()
+		}()
+	}
+	time.Sleep(20 * time.Millisecond)
+	close(release)
+	wg.Wait()
+
+	require.Equal(t, int64(1), calls.Load(), "concurrent misses collapse into one fetch")
+}
+
+func TestStatsCache_ErrorNotCached(t *testing.T) {
+	var calls atomic.Int64
+	wantErr := errors.New("dump failed")
+	c := newStatsCache(time.Minute, func() (map[string]WGStats, error) {
+		calls.Add(1)
+		return nil, wantErr
+	})
+
+	_, err := c.get()
+	require.ErrorIs(t, err, wantErr)
+	_, err = c.get()
+	require.ErrorIs(t, err, wantErr)
+	require.Equal(t, int64(2), calls.Load(), "errors are not cached; each call retries")
+}

client/iface/configurer/usp.go

@@ -40,6 +40,7 @@ type WGUSPConfigurer struct {
 	device           *device.Device
 	deviceName       string
 	activityRecorder *bind.ActivityRecorder
+	statsCache       *statsCache
 
 	uapiListener net.Listener
 }
@@ -50,16 +51,19 @@ func NewUSPConfigurer(device *device.Device, deviceName string, activityRecorder
 		deviceName:       deviceName,
 		activityRecorder: activityRecorder,
 	}
+	wgCfg.statsCache = newStatsCache(statsCacheTTL, wgCfg.fetchStats)
 	wgCfg.startUAPI()
 	return wgCfg
 }
 
 func NewUSPConfigurerNoUAPI(device *device.Device, deviceName string, activityRecorder *bind.ActivityRecorder) *WGUSPConfigurer {
-	return &WGUSPConfigurer{
+	wgCfg := &WGUSPConfigurer{
 		device:           device,
 		deviceName:       deviceName,
 		activityRecorder: activityRecorder,
 	}
+	wgCfg.statsCache = newStatsCache(statsCacheTTL, wgCfg.fetchStats)
+	return wgCfg
 }
 
 func (c *WGUSPConfigurer) ConfigureInterface(privateKey string, port int) error {
@@ -348,6 +352,10 @@ func (t *WGUSPConfigurer) Close() {
 }
 
 func (t *WGUSPConfigurer) GetStats() (map[string]WGStats, error) {
+	return t.statsCache.get()
+}
+
+func (t *WGUSPConfigurer) fetchStats() (map[string]WGStats, error) {
 	ipc, err := t.device.IpcGet()
 	if err != nil {
 		return nil, fmt.Errorf("ipc get: %w", err)

client/internal/auth/auth.go

@@ -351,7 +351,6 @@ func (a *Auth) setSystemInfoFlags(info *system.Info) {
 		a.config.BlockLANAccess,
 		a.config.BlockInbound,
 		a.config.DisableIPv6,
-		a.config.LazyConnectionEnabled,
 		a.config.EnableSSHRoot,
 		a.config.EnableSSHSFTP,
 		a.config.EnableSSHLocalPortForwarding,

client/internal/auth/device_flow.go

@@ -259,12 +259,18 @@ func (d *DeviceAuthorizationFlow) WaitToken(ctx context.Context, info AuthFlowIn
 	ticker := time.NewTicker(interval)
 	defer ticker.Stop()
 
+	log.Infof("device flow: waiting for user authorization, polling token endpoint every %s, code expires in %s", interval, timeout)
+
+	start := time.Now()
+	polls := 0
+
 	for {
 		select {
 		case <-waitCtx.Done():
 			return TokenInfo{}, waitCtx.Err()
 		case <-ticker.C:
 
+			polls++
 			tokenResponse, err := d.requestToken(info)
 			if err != nil {
 				return TokenInfo{}, fmt.Errorf("parsing token response failed with error: %v", err)
@@ -272,10 +278,12 @@ func (d *DeviceAuthorizationFlow) WaitToken(ctx context.Context, info AuthFlowIn
 
 			if tokenResponse.Error != "" {
 				if tokenResponse.Error == "authorization_pending" {
+					log.Tracef("device flow: authorization still pending after poll %d", polls)
 					continue
 				} else if tokenResponse.Error == "slow_down" {
 					interval += (3 * time.Second)
 					ticker.Reset(interval)
+					log.Infof("device flow: IdP requested slow_down, polling interval increased to %s", interval)
 					continue
 				}
 
@@ -296,6 +304,7 @@ func (d *DeviceAuthorizationFlow) WaitToken(ctx context.Context, info AuthFlowIn
 				return TokenInfo{}, fmt.Errorf("validate access token failed with error: %v", err)
 			}
 
+			log.Infof("device flow: user authorization confirmed after %d polls in %s", polls, time.Since(start).Round(time.Second))
 			return tokenInfo, err
 		}
 	}

client/internal/auth/pkce_flow.go

@@ -188,6 +188,8 @@ func (p *PKCEAuthorizationFlow) WaitToken(ctx context.Context, info AuthFlowInfo
 	waitCtx, cancel := context.WithTimeout(ctx, timeout)
 	defer cancel()
 
+	log.Infof("pkce flow: waiting for authorization callback on %s, timeout %s", p.oAuthConfig.RedirectURL, timeout)
+
 	tokenChan := make(chan *oauth2.Token, 1)
 	errChan := make(chan error, 1)
 
@@ -221,6 +223,7 @@ func (p *PKCEAuthorizationFlow) WaitToken(ctx context.Context, info AuthFlowInfo
 func (p *PKCEAuthorizationFlow) startServer(server *http.Server, tokenChan chan<- *oauth2.Token, errChan chan<- error) {
 	mux := http.NewServeMux()
 	mux.HandleFunc("/", func(w http.ResponseWriter, req *http.Request) {
+		log.Infof("pkce flow: received authorization callback from IdP")
 		cert := p.providerConfig.ClientCertPair
 		if cert != nil {
 			tr := &http.Transport{
@@ -271,11 +274,18 @@ func (p *PKCEAuthorizationFlow) handleRequest(req *http.Request) (*oauth2.Token,
 		return nil, fmt.Errorf("authentication failed: missing code")
 	}
 
-	return p.oAuthConfig.Exchange(
+	exchangeStart := time.Now()
+	token, err := p.oAuthConfig.Exchange(
 		req.Context(),
 		code,
 		oauth2.SetAuthURLParam("code_verifier", p.codeVerifier),
 	)
+	if err != nil {
+		return nil, err
+	}
+
+	log.Infof("pkce flow: authorization code exchanged for token in %s", time.Since(exchangeStart).Round(time.Millisecond))
+	return token, nil
 }
 
 func (p *PKCEAuthorizationFlow) parseOAuthToken(token *oauth2.Token) (TokenInfo, error) {

client/internal/conn_mgr.go

@@ -16,6 +16,16 @@ import (
 	"github.com/netbirdio/netbird/route"
 )
 
+// lazyForce is the resolved local decision for lazy connections, layered above the
+// management feature flag. lazyForceNone defers to management.
+type lazyForce int
+
+const (
+	lazyForceNone lazyForce = iota
+	lazyForceOn
+	lazyForceOff
+)
+
 // ConnMgr coordinates both lazy connections (established on-demand) and permanent peer connections.
 //
 // The connection manager is responsible for:
@@ -28,7 +38,7 @@ type ConnMgr struct {
 	peerStore        *peerstore.Store
 	statusRecorder   *peer.Status
 	iface            lazyconn.WGIface
-	enabledLocally   bool
+	force            lazyForce
 	rosenpassEnabled bool
 
 	lazyConnMgr *manager.Manager
@@ -43,28 +53,34 @@ func NewConnMgr(engineConfig *EngineConfig, statusRecorder *peer.Status, peerSto
 		peerStore:        peerStore,
 		statusRecorder:   statusRecorder,
 		iface:            iface,
+		force:            resolveLazyForce(engineConfig.LazyConnection),
 		rosenpassEnabled: engineConfig.RosenpassEnabled,
 	}
-	if engineConfig.LazyConnectionEnabled || lazyconn.IsLazyConnEnabledByEnv() {
-		e.enabledLocally = true
-	}
 	return e
 }
 
-// Start initializes the connection manager and starts the lazy connection manager if enabled by env var or cmd line option.
+// Start initializes the connection manager. It starts the lazy connection manager when a
+// local override forces it on; with no local override it waits for the management feature flag.
 func (e *ConnMgr) Start(ctx context.Context) {
 	if e.lazyConnMgr != nil {
 		log.Errorf("lazy connection manager is already started")
 		return
 	}
 
-	if !e.enabledLocally {
-		log.Infof("lazy connection manager is disabled")
+	switch e.force {
+	case lazyForceOff:
+		log.Infof("lazy connection manager is disabled by local override (%s or MDM policy)", lazyconn.EnvLazyConn)
+		e.statusRecorder.UpdateLazyConnection(false)
+		return
+	case lazyForceNone:
+		log.Infof("lazy connection manager is managed by the management feature flag")
+		e.statusRecorder.UpdateLazyConnection(false)
 		return
 	}
 
 	if e.rosenpassEnabled {
 		log.Warnf("rosenpass connection manager is enabled, lazy connection manager will not be started")
+		e.statusRecorder.UpdateLazyConnection(false)
 		return
 	}
 
@@ -76,8 +92,8 @@ func (e *ConnMgr) Start(ctx context.Context) {
 // If enabled, it initializes the lazy connection manager and start it. Do not need to call Start() again.
 // If disabled, then it closes the lazy connection manager and open the connections to all peers.
 func (e *ConnMgr) UpdatedRemoteFeatureFlag(ctx context.Context, enabled bool) error {
-	// do not disable lazy connection manager if it was enabled by env var
-	if e.enabledLocally {
+	// a local override (NB_LAZY_CONN or local config) takes precedence over management
+	if e.force != lazyForceNone {
 		return nil
 	}
 
@@ -89,15 +105,17 @@ func (e *ConnMgr) UpdatedRemoteFeatureFlag(ctx context.Context, enabled bool) er
 
 		if e.rosenpassEnabled {
 			log.Infof("rosenpass connection manager is enabled, lazy connection manager will not be started")
+			e.statusRecorder.UpdateLazyConnection(false)
 			return nil
 		}
 
-		log.Warnf("lazy connection manager is enabled by management feature flag")
+		log.Infof("lazy connection manager is enabled by the management feature flag")
 		e.initLazyManager(ctx)
 		e.statusRecorder.UpdateLazyConnection(true)
 		return e.addPeersToLazyConnManager()
 	} else {
 		if e.lazyConnMgr == nil {
+			e.statusRecorder.UpdateLazyConnection(false)
 			return nil
 		}
 		log.Infof("lazy connection manager is disabled by management feature flag")
@@ -107,37 +125,6 @@ func (e *ConnMgr) UpdatedRemoteFeatureFlag(ctx context.Context, enabled bool) er
 	}
 }
 
-// SetLocalLazyConn applies a local lazy connection override (UI / CLI / env).
-// While enabledLocally is true, UpdatedRemoteFeatureFlag (management sync) is a
-// no-op, so the local setting wins until it is turned off again.
-func (e *ConnMgr) SetLocalLazyConn(ctx context.Context, enabled bool) error {
-	e.enabledLocally = enabled
-
-	if enabled {
-		if e.lazyConnMgr != nil {
-			return nil
-		}
-
-		if e.rosenpassEnabled {
-			log.Warnf("rosenpass connection manager is enabled, lazy connection manager will not be started")
-			return nil
-		}
-
-		log.Infof("lazy connection manager is enabled locally")
-		e.initLazyManager(ctx)
-		e.statusRecorder.UpdateLazyConnection(true)
-		return e.addPeersToLazyConnManager()
-	}
-
-	if e.lazyConnMgr == nil {
-		return nil
-	}
-	log.Infof("lazy connection manager is disabled locally")
-	e.closeManager(ctx)
-	e.statusRecorder.UpdateLazyConnection(false)
-	return nil
-}
-
 // UpdateRouteHAMap updates the route HA mappings in the lazy connection manager
 func (e *ConnMgr) UpdateRouteHAMap(haMap route.HAMap) {
 	if !e.isStartedWithLazyMgr() {
@@ -340,6 +327,25 @@ func (e *ConnMgr) isStartedWithLazyMgr() bool {
 	return e.lazyConnMgr != nil && e.lazyCtxCancel != nil
 }
 
+// resolveLazyForce determines the local override. NB_LAZY_CONN takes precedence; when it
+// is unset the MDM policy override (mdmState) applies. Either wins in both directions over
+// the management feature flag; StateUnset for both defers to management.
+func resolveLazyForce(mdmState lazyconn.State) lazyForce {
+	state := lazyconn.EnvState()
+	if state == lazyconn.StateUnset {
+		state = mdmState
+	}
+
+	switch state {
+	case lazyconn.StateOn:
+		return lazyForceOn
+	case lazyconn.StateOff:
+		return lazyForceOff
+	default:
+		return lazyForceNone
+	}
+}
+
 func inactivityThresholdEnv() *time.Duration {
 	envValue := os.Getenv(lazyconn.EnvInactivityThreshold)
 	if envValue == "" {

client/internal/conn_mgr_test.go

@@ -0,0 +1,40 @@
+package internal
+
+import (
+	"os"
+	"testing"
+
+	"github.com/netbirdio/netbird/client/internal/lazyconn"
+)
+
+func TestResolveLazyForce(t *testing.T) {
+	tests := []struct {
+		name   string
+		env    string
+		envSet bool
+		mdm    lazyconn.State
+		want   lazyForce
+	}{
+		{name: "env unset, mdm unset -> defer to management", mdm: lazyconn.StateUnset, want: lazyForceNone},
+		{name: "env on -> force on", env: "on", envSet: true, mdm: lazyconn.StateUnset, want: lazyForceOn},
+		{name: "env off -> force off", env: "off", envSet: true, mdm: lazyconn.StateUnset, want: lazyForceOff},
+		{name: "env unset, mdm on -> force on", mdm: lazyconn.StateOn, want: lazyForceOn},
+		{name: "env unset, mdm off -> force off", mdm: lazyconn.StateOff, want: lazyForceOff},
+		{name: "env on beats mdm off", env: "on", envSet: true, mdm: lazyconn.StateOff, want: lazyForceOn},
+		{name: "env off beats mdm on", env: "off", envSet: true, mdm: lazyconn.StateOn, want: lazyForceOff},
+		{name: "unrecognized env, mdm on -> mdm wins", env: "auto", envSet: true, mdm: lazyconn.StateOn, want: lazyForceOn},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Setenv(lazyconn.EnvLazyConn, tt.env)
+			if !tt.envSet {
+				os.Unsetenv(lazyconn.EnvLazyConn)
+			}
+
+			if got := resolveLazyForce(tt.mdm); got != tt.want {
+				t.Fatalf("resolveLazyForce(%v) = %v, want %v", tt.mdm, got, tt.want)
+			}
+		})
+	}
+}

client/internal/connect.go

@@ -27,6 +27,7 @@ import (
 	"github.com/netbirdio/netbird/client/iface/device"
 	"github.com/netbirdio/netbird/client/iface/netstack"
 	"github.com/netbirdio/netbird/client/internal/dns"
+	"github.com/netbirdio/netbird/client/internal/lazyconn"
 	"github.com/netbirdio/netbird/client/internal/listener"
 	"github.com/netbirdio/netbird/client/internal/metrics"
 	"github.com/netbirdio/netbird/client/internal/peer"
@@ -618,7 +619,7 @@ func createEngineConfig(key wgtypes.Key, config *profilemanager.Config, peerConf
 		BlockInbound:        config.BlockInbound,
 		DisableIPv6:         config.DisableIPv6,
 
-		LazyConnectionEnabled: config.LazyConnectionEnabled,
+		LazyConnection: lazyconn.ParseState(config.LazyConnection),
 
 		MTU:     selectMTU(config.MTU, peerConfig.Mtu),
 		LogPath: logPath,
@@ -692,7 +693,6 @@ func loginToManagement(ctx context.Context, client mgm.Client, pubSSHKey []byte,
 		config.BlockLANAccess,
 		config.BlockInbound,
 		config.DisableIPv6,
-		config.LazyConnectionEnabled,
 		config.EnableSSHRoot,
 		config.EnableSSHSFTP,
 		config.EnableSSHLocalPortForwarding,

client/internal/debug/debug.go

@@ -695,7 +695,7 @@ func (g *BundleGenerator) addCommonConfigFields(configContent *strings.Builder)
 		configContent.WriteString(fmt.Sprintf("ClientCertKeyPath: %s\n", g.internalConfig.ClientCertKeyPath))
 	}
 
-	configContent.WriteString(fmt.Sprintf("LazyConnectionEnabled: %v\n", g.internalConfig.LazyConnectionEnabled))
+	configContent.WriteString(fmt.Sprintf("LazyConnection: %q\n", g.internalConfig.LazyConnection))
 	configContent.WriteString(fmt.Sprintf("MTU: %d\n", g.internalConfig.MTU))
 }
 

client/internal/debug/debug_test.go

@@ -885,7 +885,7 @@ func TestAddConfig_AllFieldsCovered(t *testing.T) {
 		DNSRouteInterval:              5 * time.Second,
 		ClientCertPath:                "/tmp/cert",
 		ClientCertKeyPath:             "/tmp/key",
-		LazyConnectionEnabled:         true,
+		LazyConnection:                "on",
 		MTU:                           1280,
 	}
 

client/internal/engine.go

@@ -40,6 +40,7 @@ import (
 	"github.com/netbirdio/netbird/client/internal/dnsfwd"
 	"github.com/netbirdio/netbird/client/internal/expose"
 	"github.com/netbirdio/netbird/client/internal/ingressgw"
+	"github.com/netbirdio/netbird/client/internal/lazyconn"
 	"github.com/netbirdio/netbird/client/internal/metrics"
 	"github.com/netbirdio/netbird/client/internal/netflow"
 	nftypes "github.com/netbirdio/netbird/client/internal/netflow/types"
@@ -47,6 +48,7 @@ import (
 	"github.com/netbirdio/netbird/client/internal/peer"
 	"github.com/netbirdio/netbird/client/internal/peer/guard"
 	icemaker "github.com/netbirdio/netbird/client/internal/peer/ice"
+	"github.com/netbirdio/netbird/client/internal/peer/signaling"
 	"github.com/netbirdio/netbird/client/internal/peerstore"
 	"github.com/netbirdio/netbird/client/internal/portforward"
 	"github.com/netbirdio/netbird/client/internal/profilemanager"
@@ -147,7 +149,9 @@ type EngineConfig struct {
 	BlockInbound        bool
 	DisableIPv6         bool
 
-	LazyConnectionEnabled bool
+	// LazyConnection is the MDM-sourced lazy-connection override; StateUnset defers to
+	// the env var and management feature flag.
+	LazyConnection lazyconn.State
 
 	MTU uint16
 
@@ -179,7 +183,7 @@ type EngineServices struct {
 type Engine struct {
 	// signal is a Signal Service client
 	signal   signal.Client
-	signaler *peer.Signaler
+	signaler *signaling.Signaler
 	// mgmClient is a Management Service client
 	mgmClient mgm.Client
 	// peerConns is a map that holds all the peers that are known to this peer
@@ -315,7 +319,7 @@ func NewEngine(
 		ctx:                ctx,
 		cancel:             cancel,
 		signal:             services.SignalClient,
-		signaler:           peer.NewSignaler(services.SignalClient, config.WgPrivateKey),
+		signaler:           signaling.NewSignaler(services.SignalClient, config.WgPrivateKey),
 		mgmClient:          services.MgmClient,
 		relayManager:       services.RelayManager,
 		peerStore:          peerstore.NewConnStore(),
@@ -548,7 +552,7 @@ func (e *Engine) Start(netbirdConfig *mgmProto.NetbirdConfig, mgmtURL *url.URL)
 		} else {
 			log.Infof("running rosenpass in strict mode")
 		}
-		e.rpManager, err = rosenpass.NewManager(e.config.PreSharedKey, e.config.WgIfaceName)
+		e.rpManager, err = rosenpass.NewManager(e.config.PreSharedKey, e.config.WgIfaceName, publicKey)
 		if err != nil {
 			return fmt.Errorf("create rosenpass manager: %w", err)
 		}
@@ -1161,7 +1165,6 @@ func (e *Engine) applyInfoFlags(info *system.Info) {
 		e.config.BlockLANAccess,
 		e.config.BlockInbound,
 		e.config.DisableIPv6,
-		e.config.LazyConnectionEnabled,
 		e.config.EnableSSHRoot,
 		e.config.EnableSSHSFTP,
 		e.config.EnableSSHLocalPortForwarding,
@@ -2030,7 +2033,6 @@ func (e *Engine) readInitialSettings() ([]*route.Route, *nbdns.Config, bool, err
 		e.config.BlockLANAccess,
 		e.config.BlockInbound,
 		e.config.DisableIPv6,
-		e.config.LazyConnectionEnabled,
 		e.config.EnableSSHRoot,
 		e.config.EnableSSHSFTP,
 		e.config.EnableSSHLocalPortForwarding,
@@ -2746,7 +2748,7 @@ func createFile(path string) error {
 	return file.Close()
 }
 
-func convertToOfferAnswer(msg *sProto.Message) (*peer.OfferAnswer, error) {
+func convertToOfferAnswer(msg *sProto.Message) (*signaling.OfferAnswer, error) {
 	remoteCred, err := signal.UnMarshalCredential(msg)
 	if err != nil {
 		return nil, err
@@ -2762,9 +2764,9 @@ func convertToOfferAnswer(msg *sProto.Message) (*peer.OfferAnswer, error) {
 	}
 
 	// Handle optional SessionID
-	var sessionID *peer.ICESessionID
+	var sessionID *icemaker.SessionID
 	if sessionBytes := msg.GetBody().GetSessionId(); sessionBytes != nil {
-		if id, err := peer.ICESessionIDFromBytes(sessionBytes); err != nil {
+		if id, err := icemaker.SessionIDFromBytes(sessionBytes); err != nil {
 			log.Warnf("Invalid session ID in message: %v", err)
 			sessionID = nil // Set to nil if conversion fails
 		} else {
@@ -2774,8 +2776,8 @@ func convertToOfferAnswer(msg *sProto.Message) (*peer.OfferAnswer, error) {
 
 	relayIP := decodeRelayIP(msg.GetBody().GetRelayServerIP())
 
-	offerAnswer := peer.OfferAnswer{
-		IceCredentials: peer.IceCredentials{
+	offerAnswer := signaling.OfferAnswer{
+		IceCredentials: signaling.IceCredentials{
 			UFrag: remoteCred.UFrag,
 			Pwd:   remoteCred.Pwd,
 		},

client/internal/engine_lazyconn.go

@@ -1,19 +0,0 @@
-package internal
-
-import (
-	"errors"
-)
-
-// SetLazyConnEnabled applies a local lazy connection override to the running
-// engine. It pins the setting like an env/CLI flag, so a later management sync
-// cannot override it. syncMsgMux guards ConnMgr, which is not thread-safe.
-func (e *Engine) SetLazyConnEnabled(enabled bool) error {
-	e.syncMsgMux.Lock()
-	defer e.syncMsgMux.Unlock()
-
-	if e.connMgr == nil {
-		return errors.New("connection manager is not initialised")
-	}
-
-	return e.connMgr.SetLocalLazyConn(e.ctx, enabled)
-}

client/internal/lazyconn/env.go

@@ -3,24 +3,57 @@ package lazyconn
 import (
 	"os"
 	"strconv"
+	"strings"
 
 	log "github.com/sirupsen/logrus"
 )
 
 const (
-	EnvEnableLazyConn      = "NB_ENABLE_EXPERIMENTAL_LAZY_CONN"
+	EnvLazyConn            = "NB_LAZY_CONN"
 	EnvInactivityThreshold = "NB_LAZY_CONN_INACTIVITY_THRESHOLD"
 )
 
-func IsLazyConnEnabledByEnv() bool {
-	val := os.Getenv(EnvEnableLazyConn)
-	if val == "" {
-		return false
-	}
-	enabled, err := strconv.ParseBool(val)
-	if err != nil {
-		log.Warnf("failed to parse %s: %v", EnvEnableLazyConn, err)
-		return false
-	}
-	return enabled
+// State is the tri-state local override for lazy connections read from the environment.
+type State int
+
+const (
+	// StateUnset means no local override; defer to the management feature flag.
+	StateUnset State = iota
+	// StateOn forces lazy connections on, overriding management.
+	StateOn
+	// StateOff forces lazy connections off, overriding management.
+	StateOff
+)
+
+// EnvState reads NB_LAZY_CONN and returns the local override state.
+func EnvState() State {
+	return ParseState(os.Getenv(EnvLazyConn))
+}
+
+// ParseState interprets a lazy-connection override value (from the environment or an MDM
+// policy). It accepts the on/off aliases plus any value strconv.ParseBool understands
+// (true/false/1/0). An empty or unrecognized value returns StateUnset so that the
+// management feature flag remains in control.
+func ParseState(raw string) State {
+	if raw == "" {
+		return StateUnset
+	}
+
+	normalized := strings.ToLower(strings.TrimSpace(raw))
+	switch normalized {
+	case "on":
+		return StateOn
+	case "off":
+		return StateOff
+	}
+
+	enabled, err := strconv.ParseBool(normalized)
+	if err != nil {
+		log.Warnf("failed to parse lazy connection value %q (from %s env or MDM policy): %v", raw, EnvLazyConn, err)
+		return StateUnset
+	}
+	if enabled {
+		return StateOn
+	}
+	return StateOff
 }

client/internal/lazyconn/env_test.go

@@ -0,0 +1,45 @@
+package lazyconn
+
+import (
+	"os"
+	"testing"
+)
+
+func TestEnvState(t *testing.T) {
+	tests := []struct {
+		value string
+		set   bool
+		want  State
+	}{
+		{set: false, want: StateUnset},
+		{value: "", set: true, want: StateUnset},
+		{value: "on", set: true, want: StateOn},
+		{value: "ON", set: true, want: StateOn},
+		{value: "true", set: true, want: StateOn},
+		{value: "1", set: true, want: StateOn},
+		{value: " on ", set: true, want: StateOn},
+		{value: "off", set: true, want: StateOff},
+		{value: "OFF", set: true, want: StateOff},
+		{value: "false", set: true, want: StateOff},
+		{value: "0", set: true, want: StateOff},
+		{value: "auto", set: true, want: StateUnset},
+		{value: "garbage", set: true, want: StateUnset},
+	}
+
+	for _, tt := range tests {
+		name := tt.value
+		if !tt.set {
+			name = "unset"
+		}
+		t.Run(name, func(t *testing.T) {
+			t.Setenv(EnvLazyConn, tt.value)
+			if !tt.set {
+				os.Unsetenv(EnvLazyConn)
+			}
+
+			if got := EnvState(); got != tt.want {
+				t.Fatalf("EnvState() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}

client/internal/netflow/logger/logger.go

@@ -27,7 +27,7 @@ type Logger struct {
 	wgIfaceNetV6       netip.Prefix
 	dnsCollection      atomic.Bool
 	exitNodeCollection atomic.Bool
-	Store              types.Store
+	Store              types.AggregatingStore
 }
 
 func New(statusRecorder *peer.Status, wgIfaceIPNet, wgIfaceIPNetV6 netip.Prefix) *Logger {
@@ -35,7 +35,7 @@ func New(statusRecorder *peer.Status, wgIfaceIPNet, wgIfaceIPNetV6 netip.Prefix)
 		statusRecorder: statusRecorder,
 		wgIfaceNet:     wgIfaceIPNet,
 		wgIfaceNetV6:   wgIfaceIPNetV6,
-		Store:          store.NewMemoryStore(),
+		Store:          store.NewAggregatingMemoryStore(),
 	}
 }
 
@@ -125,6 +125,10 @@ func (l *Logger) stop() {
 	l.mux.Unlock()
 }
 
+func (l *Logger) ResetAggregationWindow() types.FlowEventAggregator {
+	return l.Store.ResetAggregationWindow()
+}
+
 func (l *Logger) GetEvents() []*types.Event {
 	return l.Store.GetEvents()
 }

client/internal/netflow/manager.go

@@ -9,12 +9,14 @@ import (
 	"sync"
 	"time"
 
+	"github.com/cenkalti/backoff/v4"
 	"github.com/google/uuid"
 	log "github.com/sirupsen/logrus"
 	"google.golang.org/protobuf/types/known/timestamppb"
 
 	"github.com/netbirdio/netbird/client/internal/netflow/conntrack"
 	"github.com/netbirdio/netbird/client/internal/netflow/logger"
+	"github.com/netbirdio/netbird/client/internal/netflow/store"
 	nftypes "github.com/netbirdio/netbird/client/internal/netflow/types"
 	"github.com/netbirdio/netbird/client/internal/peer"
 	"github.com/netbirdio/netbird/flow/client"
@@ -23,14 +25,16 @@ import (
 
 // Manager handles netflow tracking and logging
 type Manager struct {
-	mux            sync.Mutex
-	shutdownWg     sync.WaitGroup
-	logger         nftypes.FlowLogger
-	flowConfig     *nftypes.FlowConfig
-	conntrack      nftypes.ConnTracker
-	receiverClient *client.GRPCClient
-	publicKey      []byte
-	cancel         context.CancelFunc
+	mux               sync.Mutex
+	shutdownWg        sync.WaitGroup
+	logger            nftypes.FlowLogger
+	flowConfig        *nftypes.FlowConfig
+	conntrack         nftypes.ConnTracker
+	receiverClient    *client.GRPCClient
+	eventsWithoutAcks nftypes.Store
+	publicKey         []byte
+	cancel            context.CancelFunc
+	retryInterval     time.Duration
 }
 
 // NewManager creates a new netflow manager
@@ -48,9 +52,11 @@ func NewManager(iface nftypes.IFaceMapper, publicKey []byte, statusRecorder *pee
 	}
 
 	return &Manager{
-		logger:    flowLogger,
-		conntrack: ct,
-		publicKey: publicKey,
+		logger:            flowLogger,
+		conntrack:         ct,
+		publicKey:         publicKey,
+		retryInterval:     time.Second,
+		eventsWithoutAcks: store.NewMemoryStore(),
 	}
 }
 
@@ -66,6 +72,7 @@ func (m *Manager) needsNewClient(previous *nftypes.FlowConfig) bool {
 }
 
 // enableFlow starts components for flow tracking
+// must be called under m.mux lock
 func (m *Manager) enableFlow(previous *nftypes.FlowConfig) error {
 	// first make sender ready so events don't pile up
 	if m.needsNewClient(previous) {
@@ -85,6 +92,7 @@ func (m *Manager) enableFlow(previous *nftypes.FlowConfig) error {
 	return nil
 }
 
+// must be called under m.mux lock
 func (m *Manager) resetClient() error {
 	if m.receiverClient != nil {
 		if err := m.receiverClient.Close(); err != nil {
@@ -107,14 +115,19 @@ func (m *Manager) resetClient() error {
 	ctx, cancel := context.WithCancel(context.Background())
 	m.cancel = cancel
 
-	m.shutdownWg.Add(2)
+	m.shutdownWg.Add(3)
+	flowConfigInterval := m.flowConfig.Interval
 	go func() {
 		defer m.shutdownWg.Done()
-		m.receiveACKs(ctx, flowClient)
+		m.receiveACKs(ctx, flowClient, flowConfigInterval)
 	}()
 	go func() {
 		defer m.shutdownWg.Done()
-		m.startSender(ctx)
+		m.startSender(ctx, flowConfigInterval)
+	}()
+	go func() {
+		defer m.shutdownWg.Done()
+		m.startRetries(ctx, flowConfigInterval)
 	}()
 
 	return nil
@@ -198,8 +211,8 @@ func (m *Manager) GetLogger() nftypes.FlowLogger {
 	return m.logger
 }
 
-func (m *Manager) startSender(ctx context.Context) {
-	ticker := time.NewTicker(m.flowConfig.Interval)
+func (m *Manager) startSender(ctx context.Context, flowConfigInterval time.Duration) {
+	ticker := time.NewTicker(flowConfigInterval)
 	defer ticker.Stop()
 
 	for {
@@ -207,27 +220,29 @@ func (m *Manager) startSender(ctx context.Context) {
 		case <-ctx.Done():
 			return
 		case <-ticker.C:
-			events := m.logger.GetEvents()
+			collectedEvents := m.logger.ResetAggregationWindow()
+			events := collectedEvents.GetAggregatedEvents()
 			for _, event := range events {
+				m.eventsWithoutAcks.StoreEvent(event)
 				if err := m.send(event); err != nil {
 					log.Errorf("failed to send flow event to server: %v", err)
-					continue
+				} else {
+					log.Tracef("sent flow event: %s", event.ID)
 				}
-				log.Tracef("sent flow event: %s", event.ID)
 			}
 		}
 	}
 }
 
-func (m *Manager) receiveACKs(ctx context.Context, client *client.GRPCClient) {
-	err := client.Receive(ctx, m.flowConfig.Interval, func(ack *proto.FlowEventAck) error {
+func (m *Manager) receiveACKs(ctx context.Context, client *client.GRPCClient, flowConfigInterval time.Duration) {
+	err := client.Receive(ctx, flowConfigInterval, func(ack *proto.FlowEventAck) error {
 		id, err := uuid.FromBytes(ack.EventId)
 		if err != nil {
 			log.Warnf("failed to convert ack event id to uuid: %v", err)
 			return nil
 		}
 		log.Tracef("received flow event ack: %s", id)
-		m.logger.DeleteEvents([]uuid.UUID{id})
+		m.eventsWithoutAcks.DeleteEvents([]uuid.UUID{id})
 		return nil
 	})
 
@@ -236,6 +251,51 @@ func (m *Manager) receiveACKs(ctx context.Context, client *client.GRPCClient) {
 	}
 }
 
+// We effectively never drop events (see MaxInterval), which makes eventsWithoutAcks unbounded.
+// We may want to limit the max size of the store, and start dropping oldest events when the threshold is reached.
+func (m *Manager) startRetries(ctx context.Context, flowConfigInterval time.Duration) {
+	timer := time.NewTimer(m.retryInterval)
+	retryBackoff := backoff.WithContext(&backoff.ExponentialBackOff{
+		InitialInterval:     1 * time.Second,
+		RandomizationFactor: 0.5,
+		Multiplier:          1.7,
+		MaxInterval:         flowConfigInterval / 2,
+		MaxElapsedTime:      3 * 30 * 24 * time.Hour, // 3 months
+		Stop:                backoff.Stop,
+		Clock:               backoff.SystemClock,
+	}, ctx)
+	defer timer.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-timer.C:
+			resetBackoff := true
+			for _, e := range m.eventsWithoutAcks.GetEvents() {
+				if e.Timestamp.Add(time.Second).After(time.Now()) {
+					// grace period on retries to avoid early retries
+					// do not retry if the event is less than 1 sec old
+					continue
+				}
+				if err := m.send(e); err != nil {
+					if nextBackoff := retryBackoff.NextBackOff(); nextBackoff != backoff.Stop {
+						timer = time.NewTimer(nextBackoff)
+						resetBackoff = false
+					} else {
+						resetBackoff = true // we exhausted retries, reset retry loop
+					}
+					break
+				}
+			}
+			if resetBackoff { // use regular retry interval in absence of network errors
+				retryBackoff.Reset()
+				timer = time.NewTimer(m.retryInterval)
+			}
+		}
+	}
+}
+
 func (m *Manager) send(event *nftypes.Event) error {
 	m.mux.Lock()
 	client := m.receiverClient
@@ -250,9 +310,11 @@ func (m *Manager) send(event *nftypes.Event) error {
 
 func toProtoEvent(publicKey []byte, event *nftypes.Event) *proto.FlowEvent {
 	protoEvent := &proto.FlowEvent{
-		EventId:   event.ID[:],
-		Timestamp: timestamppb.New(event.Timestamp),
-		PublicKey: publicKey,
+		EventId:     event.ID[:],
+		Timestamp:   timestamppb.New(event.Timestamp),
+		PublicKey:   publicKey,
+		WindowStart: timestamppb.New(event.WindowStart),
+		WindowEnd:   timestamppb.New(event.WindowEnd),
 		FlowFields: &proto.FlowFields{
 			FlowId:           event.FlowID[:],
 			RuleId:           event.RuleID,
@@ -267,6 +329,9 @@ func toProtoEvent(publicKey []byte, event *nftypes.Event) *proto.FlowEvent {
 			TxBytes:          event.TxBytes,
 			SourceResourceId: event.SourceResourceID,
 			DestResourceId:   event.DestResourceID,
+			NumOfStarts:      event.NumOfStarts,
+			NumOfEnds:        event.NumOfEnds,
+			NumOfDrops:       event.NumOfDrops,
 		},
 	}
 

client/internal/netflow/manager_integration_test.go

@@ -0,0 +1,291 @@
+package netflow
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net"
+	"net/netip"
+	"slices"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/netbirdio/netbird/client/iface/wgaddr"
+	"github.com/netbirdio/netbird/client/internal/netflow/types"
+	"github.com/netbirdio/netbird/flow/proto"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"google.golang.org/grpc"
+)
+
+type testServer struct {
+	proto.UnimplementedFlowServiceServer
+	events         chan *proto.FlowEvent
+	acks           chan *proto.FlowEventAck
+	grpcSrv        *grpc.Server
+	addr           string
+	handlerDone    chan struct{} // signaled each time Events() exits
+	handlerStarted chan struct{} // signaled each time Events() begins
+}
+
+func newTestServer(t *testing.T) *testServer {
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	require.NoError(t, err)
+
+	s := &testServer{
+		events:         make(chan *proto.FlowEvent, 100),
+		acks:           make(chan *proto.FlowEventAck, 100),
+		grpcSrv:        grpc.NewServer(),
+		addr:           listener.Addr().String(),
+		handlerDone:    make(chan struct{}, 10),
+		handlerStarted: make(chan struct{}, 10),
+	}
+
+	proto.RegisterFlowServiceServer(s.grpcSrv, s)
+
+	go func() {
+		if err := s.grpcSrv.Serve(listener); err != nil && !errors.Is(err, grpc.ErrServerStopped) {
+			t.Logf("server error: %v", err)
+		}
+	}()
+
+	t.Cleanup(func() {
+		s.grpcSrv.Stop()
+	})
+
+	return s
+}
+
+func (s *testServer) Events(stream proto.FlowService_EventsServer) error {
+	defer func() {
+		select {
+		case s.handlerDone <- struct{}{}:
+		default:
+		}
+	}()
+
+	err := stream.Send(&proto.FlowEventAck{IsInitiator: true})
+	if err != nil {
+		return err
+	}
+
+	select {
+	case s.handlerStarted <- struct{}{}:
+	default:
+	}
+
+	ctx, cancel := context.WithCancel(stream.Context())
+	defer cancel()
+
+	go func() {
+		defer cancel()
+		for {
+			event, err := stream.Recv()
+			if err != nil {
+				return
+			}
+
+			if !event.IsInitiator {
+				select {
+				case s.events <- event:
+				case <-ctx.Done():
+					return
+				}
+			}
+		}
+	}()
+
+	for {
+		select {
+		case ack := <-s.acks:
+			if err := stream.Send(ack); err != nil {
+				return err
+			}
+		case <-ctx.Done():
+			return ctx.Err()
+		}
+	}
+}
+
+func TestSendEventReceiveAck(t *testing.T) {
+	_, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	t.Cleanup(cancel)
+
+	server := newTestServer(t)
+	manager := createManager(t, server.addr, 60*time.Second) // set high to prevent retries in this test
+	defer manager.Close()
+
+	assert.Eventually(t, func() bool {
+		select {
+		case <-server.handlerStarted:
+			return true
+		default:
+			return false
+		}
+	}, 3*time.Second, 100*time.Millisecond)
+
+	event1 := types.EventFields{
+		FlowID:    uuid.New(),
+		Type:      types.TypeStart,
+		Direction: types.Ingress,
+		DestIP:    ipAddr("172.16.1.2"),
+		DestPort:  2345,
+		Protocol:  6,
+	}
+	manager.logger.StoreEvent(event1)
+	event2 := types.EventFields{
+		FlowID:    uuid.New(),
+		Type:      types.TypeStart,
+		Direction: types.Ingress,
+		DestIP:    ipAddr("172.16.1.1"),
+		DestPort:  1234,
+		Protocol:  6,
+	}
+	manager.logger.StoreEvent(event2)
+
+	// verify the server received logged events
+	serverSideEvents := make([]*proto.FlowEvent, 0)
+	assert.Eventually(t, func() bool {
+		select {
+		case event := <-server.events:
+			serverSideEvents = append(serverSideEvents, event)
+			if len(serverSideEvents) == 2 {
+				return true
+			}
+		default:
+			if len(serverSideEvents) == 2 {
+				return true
+			}
+		}
+		return false
+	}, 5*time.Second, 100*time.Millisecond)
+
+	serverSideFlowIds := make([]uuid.UUID, 0, 2)
+	slices.Values(serverSideEvents)(func(e *proto.FlowEvent) bool {
+		id, err := uuid.FromBytes(e.FlowFields.FlowId)
+		assert.NoError(t, err)
+		serverSideFlowIds = append(serverSideFlowIds, id)
+		return true
+	})
+	assert.ElementsMatch(t, []uuid.UUID{event1.FlowID, event2.FlowID}, serverSideFlowIds)
+
+	// verify the manager tracks un-acked events
+	unackedEvents := manager.eventsWithoutAcks.GetEvents()
+	assert.Len(t, unackedEvents, 2)
+	flowIds := make([]uuid.UUID, 0)
+	slices.Values(unackedEvents)(func(e *types.Event) bool {
+		flowIds = append(flowIds, e.FlowID)
+		return true
+	})
+	assert.ElementsMatch(t, flowIds, []uuid.UUID{event1.FlowID, event2.FlowID})
+}
+
+// verify handling of retries:
+//   - unacked events are retried
+//   - when acks arrive, events are removed from the un-acked event tracker
+func TestRetryEvents(t *testing.T) {
+	_, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	t.Cleanup(cancel)
+
+	server := newTestServer(t)
+	manager := createManager(t, server.addr, time.Second) // set low to start retries sooner
+	defer manager.Close()
+
+	assert.Eventually(t, func() bool {
+		select {
+		case <-server.handlerStarted:
+			return true
+		default:
+			return false
+		}
+	}, 3*time.Second, 100*time.Millisecond)
+
+	event1 := types.EventFields{
+		FlowID:    uuid.New(),
+		Type:      types.TypeStart,
+		Direction: types.Ingress,
+		DestIP:    ipAddr("172.16.1.2"),
+		DestPort:  2345,
+		Protocol:  6,
+	}
+	manager.logger.StoreEvent(event1)
+	event2 := types.EventFields{
+		FlowID:    uuid.New(),
+		Type:      types.TypeStart,
+		Direction: types.Ingress,
+		DestIP:    ipAddr("172.16.1.1"),
+		DestPort:  1234,
+		Protocol:  6,
+	}
+	manager.logger.StoreEvent(event2)
+
+	// verify the server received retries of logged events
+	serverSideEvents := make([]*proto.FlowEvent, 0)
+	func() {
+		c := time.After(2500 * time.Millisecond)
+		for {
+			select {
+			case event := <-server.events:
+				serverSideEvents = append(serverSideEvents, event)
+			case <-c:
+				return
+			}
+		}
+	}()
+	assert.True(t, len(serverSideEvents) > 2) // must see retries
+
+	uniqueServerSideEvents := make(map[uuid.UUID]*proto.FlowEvent)
+	slices.Values(serverSideEvents)(func(e *proto.FlowEvent) bool {
+		id, err := uuid.FromBytes(e.FlowFields.FlowId)
+		assert.NoError(t, err)
+		uniqueServerSideEvents[id] = e
+		return true
+	})
+	assert.Contains(t, uniqueServerSideEvents, event1.FlowID)
+	assert.Contains(t, uniqueServerSideEvents, event2.FlowID)
+
+	// ack events
+	server.acks <- &proto.FlowEventAck{EventId: uniqueServerSideEvents[event1.FlowID].EventId}
+	server.acks <- &proto.FlowEventAck{EventId: uniqueServerSideEvents[event2.FlowID].EventId}
+
+	assert.EventuallyWithT(t, func(c *assert.CollectT) {
+		unackedEvents := manager.eventsWithoutAcks.GetEvents()
+		assert.Empty(c, unackedEvents)
+
+	}, 3*time.Second, 100*time.Millisecond)
+}
+
+func createManager(t *testing.T, serverAddr string, retryInterval time.Duration) *Manager {
+	t.Helper()
+
+	mockIFace := &mockIFaceMapper{
+		address: wgaddr.Address{
+			Network: netip.MustParsePrefix("192.168.1.1/32"),
+		},
+		isUserspaceBind: true,
+	}
+
+	publicKey := []byte("test-public-key")
+	manager := NewManager(mockIFace, publicKey, nil)
+	manager.retryInterval = retryInterval
+
+	initialConfig := &types.FlowConfig{
+		Enabled:        true,
+		URL:            fmt.Sprintf("http://%s", serverAddr),
+		TokenPayload:   "initial-payload",
+		TokenSignature: "initial-signature",
+		Interval:       500 * time.Millisecond,
+	}
+
+	err := manager.Update(initialConfig)
+	require.NoError(t, err)
+
+	return manager
+}
+
+func ipAddr(a string) netip.Addr {
+	addr, _ := netip.ParseAddr(a)
+	return addr
+}

client/internal/netflow/store/event_aggregation_test.go

@@ -0,0 +1,365 @@
+package store
+
+import (
+	"math/rand"
+	"net/netip"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/netbirdio/netbird/client/internal/netflow/types"
+	"github.com/stretchr/testify/assert"
+)
+
+var random = rand.New(rand.NewSource(time.Now().UnixNano()))
+
+func TestFlowAggregation(t *testing.T) {
+	var protocols = []types.Protocol{types.ICMP, types.ICMPv6, types.TCP, types.UDP}
+	var tests = []struct {
+		description string
+		addresses   [][]netip.Addr
+		dstPort     uint16
+		eventTypes  []types.Type
+	}{
+		{
+			description: "start and stop",
+			addresses:   [][]netip.Addr{{netip.MustParseAddr("1.1.1.1"), netip.MustParseAddr("2.2.2.2")}, {netip.MustParseAddr("3.3.3.3"), netip.MustParseAddr("2.2.2.2")}},
+			dstPort:     uint16(random.Uint32() >> 16),
+			eventTypes:  []types.Type{types.TypeStart, types.TypeEnd},
+		},
+		{
+			description: "start and drop",
+			addresses:   [][]netip.Addr{{netip.MustParseAddr("1.1.1.1"), netip.MustParseAddr("2.2.2.2")}, {netip.MustParseAddr("3.3.3.3"), netip.MustParseAddr("2.2.2.2")}},
+			dstPort:     uint16(random.Uint32() >> 16),
+			eventTypes:  []types.Type{types.TypeStart, types.TypeDrop},
+		},
+		{
+			description: "start only",
+			addresses:   [][]netip.Addr{{netip.MustParseAddr("1.1.1.1"), netip.MustParseAddr("2.2.2.2")}, {netip.MustParseAddr("3.3.3.3"), netip.MustParseAddr("2.2.2.2")}},
+			dstPort:     uint16(random.Uint32() >> 16),
+			eventTypes:  []types.Type{types.TypeStart},
+		},
+		{
+			description: "drop only",
+			addresses:   [][]netip.Addr{{netip.MustParseAddr("1.1.1.1"), netip.MustParseAddr("2.2.2.2")}, {netip.MustParseAddr("3.3.3.3"), netip.MustParseAddr("2.2.2.2")}},
+			dstPort:     uint16(random.Uint32() >> 16),
+			eventTypes:  []types.Type{types.TypeDrop},
+		}}
+
+	for _, protocol := range protocols {
+		for _, tt := range tests {
+			t.Run(tt.description+" "+protocol.String(), func(t *testing.T) {
+				store := NewAggregatingMemoryStore()
+				store.WindowEnd = time.Now().Add(5 * time.Second)
+
+				allExpected := make([]*types.Event, 0)
+
+				for _, srcAndDst := range tt.addresses {
+					inEvents, expected := generateEvents(srcAndDst[0], srcAndDst[1], tt.dstPort, tt.eventTypes, protocol, types.Ingress, 0, store.WindowStart, store.WindowEnd)
+					for _, e := range inEvents {
+						store.StoreEvent(e)
+					}
+					allExpected = append(allExpected, expected)
+				}
+
+				events := store.GetAggregatedEvents()
+				assert.ElementsMatch(t, events, allExpected)
+			})
+		}
+	}
+}
+
+func TestIcmpEventAggregation(t *testing.T) {
+	var protocols = []types.Protocol{types.ICMP, types.ICMPv6}
+	var icmpTypes = []uint8{1, 2, 3}
+
+	var tests = []struct {
+		description string
+		addresses   [][]netip.Addr
+		eventTypes  []types.Type
+	}{
+		{
+			description: "start and stop",
+			addresses:   [][]netip.Addr{{netip.MustParseAddr("1.1.1.1"), netip.MustParseAddr("2.2.2.2")}},
+			eventTypes:  []types.Type{types.TypeStart, types.TypeEnd},
+		},
+		{
+			description: "start and drop",
+			addresses:   [][]netip.Addr{{netip.MustParseAddr("1.1.1.1"), netip.MustParseAddr("2.2.2.2")}},
+			eventTypes:  []types.Type{types.TypeStart, types.TypeDrop},
+		},
+		{
+			description: "start only",
+			addresses:   [][]netip.Addr{{netip.MustParseAddr("1.1.1.1"), netip.MustParseAddr("2.2.2.2")}},
+			eventTypes:  []types.Type{types.TypeStart},
+		},
+		{
+			description: "drop only",
+			addresses:   [][]netip.Addr{{netip.MustParseAddr("1.1.1.1"), netip.MustParseAddr("2.2.2.2")}},
+			eventTypes:  []types.Type{types.TypeDrop},
+		}}
+
+	for _, protocol := range protocols {
+		for _, tt := range tests {
+			t.Run(tt.description+" "+protocol.String(), func(t *testing.T) {
+				store := NewAggregatingMemoryStore()
+				store.WindowEnd = time.Now().Add(5 * time.Second)
+
+				allExpected := make([]*types.Event, 0)
+				for _, icmpType := range icmpTypes {
+					events, expected := generateEvents(tt.addresses[0][0], tt.addresses[0][1], 0, tt.eventTypes, protocol, types.Ingress, icmpType, store.WindowStart, store.WindowEnd)
+					for _, e := range events {
+						store.StoreEvent(e)
+					}
+					allExpected = append(allExpected, expected)
+				}
+				aggregatedEvents := store.GetAggregatedEvents()
+				assert.Len(t, aggregatedEvents, len(allExpected))
+				assert.ElementsMatch(t, aggregatedEvents, allExpected)
+			})
+		}
+	}
+}
+
+func TestFlowAggregationOfUnknownProtocols(t *testing.T) {
+	var tests = []struct {
+		description string
+		addresses   [][]netip.Addr
+		dstPort     uint16
+		eventTypes  []types.Type
+	}{
+		{
+			description: "start and stop",
+			addresses:   [][]netip.Addr{{netip.MustParseAddr("1.1.1.1"), netip.MustParseAddr("2.2.2.2")}, {netip.MustParseAddr("3.3.3.3"), netip.MustParseAddr("2.2.2.2")}},
+			dstPort:     uint16(random.Uint32() >> 16),
+			eventTypes:  []types.Type{types.TypeStart, types.TypeEnd},
+		},
+		{
+			description: "start and drop",
+			addresses:   [][]netip.Addr{{netip.MustParseAddr("1.1.1.1"), netip.MustParseAddr("2.2.2.2")}, {netip.MustParseAddr("3.3.3.3"), netip.MustParseAddr("2.2.2.2")}},
+			dstPort:     uint16(random.Uint32() >> 16),
+			eventTypes:  []types.Type{types.TypeStart, types.TypeDrop},
+		},
+		{
+			description: "start only",
+			addresses:   [][]netip.Addr{{netip.MustParseAddr("1.1.1.1"), netip.MustParseAddr("2.2.2.2")}, {netip.MustParseAddr("3.3.3.3"), netip.MustParseAddr("2.2.2.2")}},
+			dstPort:     uint16(random.Uint32() >> 16),
+			eventTypes:  []types.Type{types.TypeStart},
+		},
+		{
+			description: "drop only",
+			addresses:   [][]netip.Addr{{netip.MustParseAddr("1.1.1.1"), netip.MustParseAddr("2.2.2.2")}, {netip.MustParseAddr("3.3.3.3"), netip.MustParseAddr("2.2.2.2")}},
+			dstPort:     uint16(random.Uint32() >> 16),
+			eventTypes:  []types.Type{types.TypeDrop},
+		}}
+
+	for _, tt := range tests {
+		t.Run(tt.description+" "+types.ProtocolUnknown.String(), func(t *testing.T) {
+			store := NewAggregatingMemoryStore()
+			store.WindowEnd = time.Now().Add(5 * time.Second)
+
+			allExpected := make([]*types.Event, 0)
+
+			for _, srcAndDst := range tt.addresses {
+				inEvents, expected := generateEventsForUnknownProtocol(srcAndDst[0], srcAndDst[1], tt.dstPort, tt.eventTypes, types.ProtocolUnknown, types.Ingress, store.WindowStart, store.WindowEnd)
+				for _, e := range inEvents {
+					store.StoreEvent(e)
+				}
+				allExpected = append(allExpected, expected...)
+			}
+
+			events := store.GetAggregatedEvents()
+			assert.ElementsMatch(t, events, allExpected)
+		})
+	}
+}
+
+func TestResetAggregationWindow(t *testing.T) {
+	now := time.Now()
+	nowFunc := func() time.Time { return now }
+	store := NewAggregatingMemoryStoreWithTimeFunc(nowFunc)
+	store.StoreEvent(&types.Event{
+		ID:        uuid.New(),
+		Timestamp: time.Now(),
+		EventFields: types.EventFields{
+			FlowID:           uuid.New(),
+			Type:             types.TypeStart,
+			Protocol:         types.TCP,
+			RuleID:           []byte("rule-id-1"),
+			Direction:        types.Ingress,
+			SourceIP:         netip.MustParseAddr("1.1.1.1"),
+			SourcePort:       1234,
+			DestIP:           netip.MustParseAddr("2.2.2.2"),
+			DestPort:         5678,
+			SourceResourceID: []byte("source-resource-id"),
+			DestResourceID:   []byte("dest-resource-id"),
+			RxPackets:        random.Uint64(),
+			TxPackets:        random.Uint64(),
+			RxBytes:          random.Uint64(),
+			TxBytes:          random.Uint64(),
+		},
+	})
+
+	now = now.Add(1 * time.Second)
+	reset := store.ResetAggregationWindow()
+	previousEvents, ok := reset.(*AggregatingMemory)
+	assert.True(t, ok)
+	assert.NotEqual(t, previousEvents.WindowStart, store.WindowStart)
+	assert.Equal(t, previousEvents.WindowEnd, store.WindowStart)
+	assert.NotEmpty(t, previousEvents.events)
+	assert.Empty(t, store.events)
+}
+
+func generateEvents(srcIp, dstIp netip.Addr, dstPort uint16, eventTypes []types.Type, protocol types.Protocol,
+	direction types.Direction, icmpType uint8, windowStart, windowEnd time.Time) ([]*types.Event, *types.Event) {
+	var rxPackets, txPackets, rxBytes, txBytes uint64
+	inEvents := make([]*types.Event, 0)
+	ts := time.Now()
+	flowId := uuid.New()
+	srcPort := uint16(random.Uint32() >> 16)
+
+	for idx, eventType := range eventTypes {
+		e := &types.Event{
+			ID:        uuid.New(),
+			Timestamp: ts.Add(time.Duration(idx) * time.Second),
+			EventFields: types.EventFields{
+				FlowID:           flowId,
+				Type:             eventType,
+				Protocol:         protocol,
+				RuleID:           []byte("rule-id-1"),
+				Direction:        direction,
+				SourceIP:         srcIp,
+				SourcePort:       srcPort,
+				DestIP:           dstIp,
+				DestPort:         dstPort,
+				SourceResourceID: []byte("source-resource-id"),
+				DestResourceID:   []byte("dest-resource-id"),
+				RxPackets:        random.Uint64(),
+				TxPackets:        random.Uint64(),
+				RxBytes:          random.Uint64(),
+				TxBytes:          random.Uint64(),
+			}}
+		rxBytes += e.RxBytes
+		txBytes += e.TxBytes
+		rxPackets += e.RxPackets
+		txPackets += e.TxPackets
+		inEvents = append(inEvents, e)
+		if protocol == types.ICMP || protocol == types.ICMPv6 {
+			e.ICMPType = icmpType
+		}
+	}
+
+	var start, end, drop uint64
+	for _, eventType := range eventTypes {
+		switch eventType {
+		case types.TypeStart:
+			start += 1
+		case types.TypeDrop:
+			drop += 1
+		case types.TypeEnd:
+			end += 1
+		}
+	}
+	aggregatedEvent := &types.Event{
+		ID:          inEvents[0].ID,
+		Timestamp:   inEvents[0].Timestamp,
+		WindowStart: windowStart,
+		WindowEnd:   windowEnd,
+		EventFields: types.EventFields{
+			FlowID:           flowId,
+			Type:             types.TypeUnknown,
+			Protocol:         inEvents[0].Protocol,
+			RuleID:           []byte("rule-id-1"),
+			Direction:        inEvents[0].Direction,
+			SourceIP:         srcIp,
+			SourcePort:       srcPort,
+			DestIP:           dstIp,
+			DestPort:         dstPort,
+			SourceResourceID: []byte("source-resource-id"),
+			DestResourceID:   []byte("dest-resource-id"),
+			RxPackets:        rxPackets,
+			TxPackets:        txPackets,
+			RxBytes:          rxBytes,
+			TxBytes:          txBytes,
+			NumOfStarts:      start,
+			NumOfEnds:        end,
+			NumOfDrops:       drop,
+		}}
+	if protocol == types.ICMP || protocol == types.ICMPv6 {
+		aggregatedEvent.ICMPType = icmpType
+	}
+
+	return inEvents, aggregatedEvent
+}
+
+func generateEventsForUnknownProtocol(srcIp, dstIp netip.Addr, dstPort uint16, eventTypes []types.Type, protocol types.Protocol,
+	direction types.Direction, windowStart, windowEnd time.Time) ([]*types.Event, []*types.Event) {
+	inEvents := make([]*types.Event, 0)
+	expectedEvents := make([]*types.Event, 0)
+
+	ts := time.Now()
+	flowId := uuid.New()
+	srcPort := uint16(random.Uint32() >> 16)
+
+	for idx, eventType := range eventTypes {
+		e := &types.Event{
+			ID:        uuid.New(),
+			Timestamp: ts.Add(time.Duration(idx) * time.Second),
+			EventFields: types.EventFields{
+				FlowID:           flowId,
+				Type:             eventType,
+				Protocol:         protocol,
+				RuleID:           []byte("rule-id-1"),
+				Direction:        direction,
+				SourceIP:         srcIp,
+				SourcePort:       srcPort,
+				DestIP:           dstIp,
+				DestPort:         dstPort,
+				SourceResourceID: []byte("source-resource-id"),
+				DestResourceID:   []byte("dest-resource-id"),
+				RxPackets:        random.Uint64(),
+				TxPackets:        random.Uint64(),
+				RxBytes:          random.Uint64(),
+				TxBytes:          random.Uint64(),
+			}}
+		inEvents = append(inEvents, e)
+
+		var start, end, drop uint64
+		switch eventType {
+		case types.TypeStart:
+			start = 1
+		case types.TypeDrop:
+			drop = 1
+		case types.TypeEnd:
+			end = 1
+		}
+
+		expectedEvents = append(expectedEvents, &types.Event{
+			ID:          e.ID,
+			Timestamp:   e.Timestamp,
+			WindowStart: windowStart,
+			WindowEnd:   windowEnd,
+			EventFields: types.EventFields{
+				FlowID:           flowId,
+				Type:             types.TypeUnknown,
+				Protocol:         e.Protocol,
+				RuleID:           []byte("rule-id-1"),
+				Direction:        e.Direction,
+				SourceIP:         srcIp,
+				SourcePort:       srcPort,
+				DestIP:           dstIp,
+				DestPort:         dstPort,
+				SourceResourceID: []byte("source-resource-id"),
+				DestResourceID:   []byte("dest-resource-id"),
+				RxPackets:        e.RxPackets,
+				TxPackets:        e.TxPackets,
+				RxBytes:          e.RxBytes,
+				TxBytes:          e.TxBytes,
+				NumOfStarts:      start,
+				NumOfEnds:        end,
+				NumOfDrops:       drop,
+			}})
+	}
+
+	return inEvents, expectedEvents
+}

client/internal/netflow/store/memory.go

@@ -1,10 +1,15 @@
 package store
 
 import (
+	"maps"
+	"math/rand"
+	v2 "math/rand/v2"
+	"net/netip"
+	"slices"
 	"sync"
+	"time"
 
 	"github.com/google/uuid"
-
 	"github.com/netbirdio/netbird/client/internal/netflow/types"
 )
 
@@ -19,6 +24,14 @@ type Memory struct {
 	events map[uuid.UUID]*types.Event
 }
 
+type AggregatingMemory struct {
+	Memory
+	WindowStart time.Time
+	WindowEnd   time.Time
+	rnd         *v2.PCG
+	nowFunc     func() time.Time
+}
+
 func (m *Memory) StoreEvent(event *types.Event) {
 	m.mux.Lock()
 	defer m.mux.Unlock()
@@ -48,3 +61,104 @@ func (m *Memory) DeleteEvents(ids []uuid.UUID) {
 		delete(m.events, id)
 	}
 }
+
+func NewAggregatingMemoryStore() *AggregatingMemory {
+	return NewAggregatingMemoryStoreWithTimeFunc(defaultNowFunc)
+}
+
+// used in tests when deterministic (less random) time intervals are required
+func NewAggregatingMemoryStoreWithTimeFunc(nowFunc func() time.Time) *AggregatingMemory {
+	return &AggregatingMemory{WindowStart: nowFunc(), Memory: Memory{events: make(map[uuid.UUID]*types.Event)}, nowFunc: nowFunc, rnd: v2.NewPCG(rand.Uint64(), rand.Uint64())}
+}
+
+func (am *AggregatingMemory) ResetAggregationWindow() types.FlowEventAggregator {
+	am.mux.Lock()
+	defer am.mux.Unlock()
+
+	now := am.nowFunc()
+	toret := AggregatingMemory{WindowStart: am.WindowStart, WindowEnd: now, Memory: Memory{events: am.events}, rnd: v2.NewPCG(rand.Uint64(), rand.Uint64())}
+
+	am.events = make(map[uuid.UUID]*types.Event)
+	am.WindowStart = now
+
+	return &toret
+}
+
+type aggregationKey struct {
+	srcAddr   netip.Addr
+	destAddr  netip.Addr
+	destPort  uint16
+	direction int
+	protocol  uint8
+	icmpType  uint8
+	unique    uint64 // used to prevent aggregation on non icmp/udp/tcp events
+}
+
+func (am *AggregatingMemory) GetAggregatedEvents() []*types.Event {
+	am.mux.Lock()
+	defer am.mux.Unlock()
+
+	aggregated := make(map[aggregationKey]*types.Event)
+	for _, v := range am.events {
+		lookupKey := aggregationKey{srcAddr: v.SourceIP, destAddr: v.DestIP, destPort: v.DestPort, direction: int(v.Direction), protocol: uint8(v.Protocol), icmpType: v.ICMPType}
+		if _, ok := aggregated[lookupKey]; !ok {
+			event := v.Clone()
+
+			switch event.Type {
+			case types.TypeStart:
+				event.NumOfStarts += 1
+			case types.TypeDrop:
+				event.NumOfDrops += 1
+			case types.TypeEnd:
+				event.NumOfEnds += 1
+			}
+			event.Type = types.TypeUnknown
+
+			// Please note that ICMPCode field isn't propagated by the manager (see flow/proto/flow.pb.go, FlowFields struct)
+			// so the field value in an icmp event in the "aggregated" doesn't matter
+
+			event.WindowStart = am.WindowStart
+			event.WindowEnd = am.WindowEnd
+
+			if event.Protocol != types.ICMP && event.Protocol != types.ICMPv6 && event.Protocol != types.UDP && event.Protocol != types.TCP {
+				lookupKey.unique = am.rnd.Uint64() // to make the lookup key unique so we don't aggregate on it
+			}
+
+			aggregated[lookupKey] = event
+			continue
+		}
+
+		aggregatedEvent := aggregated[lookupKey]
+		if aggregatedEvent.Protocol != types.ICMP && aggregatedEvent.Protocol != types.ICMPv6 && aggregatedEvent.Protocol != types.UDP && aggregatedEvent.Protocol != types.TCP {
+			continue // we don't aggregate this type of events; shouldn't ever get here
+		}
+
+		// track the number of connections, duration?, open and close events?
+		aggregatedEvent.RxBytes += v.RxBytes
+		aggregatedEvent.RxPackets += v.RxPackets
+		aggregatedEvent.TxBytes += v.TxBytes
+		aggregatedEvent.TxPackets += v.TxPackets
+		switch v.Type {
+		case types.TypeStart:
+			aggregatedEvent.NumOfStarts += 1
+		case types.TypeDrop:
+			aggregatedEvent.NumOfDrops += 1
+		case types.TypeEnd:
+			aggregatedEvent.NumOfEnds += 1
+		}
+		if aggregatedEvent.Timestamp.Compare(v.Timestamp) > 0 {
+			aggregatedEvent.Timestamp = v.Timestamp
+			aggregatedEvent.ID = v.ID
+			aggregatedEvent.SourcePort = v.SourcePort
+		}
+		if len(aggregatedEvent.RuleID) == 0 && len(v.RuleID) != 0 {
+			aggregatedEvent.RuleID = slices.Clone(v.RuleID)
+		}
+	}
+
+	return slices.Collect(maps.Values(aggregated)) // could return an iterator instead here
+}
+
+func defaultNowFunc() time.Time {
+	return time.Now()
+}

client/internal/netflow/types/types.go

@@ -2,6 +2,7 @@ package types
 
 import (
 	"net/netip"
+	"slices"
 	"strconv"
 	"time"
 
@@ -69,8 +70,10 @@ const (
 )
 
 type Event struct {
-	ID        uuid.UUID
-	Timestamp time.Time
+	ID          uuid.UUID
+	Timestamp   time.Time
+	WindowStart time.Time
+	WindowEnd   time.Time
 	EventFields
 }
 
@@ -92,6 +95,17 @@ type EventFields struct {
 	TxPackets        uint64
 	RxBytes          uint64
 	TxBytes          uint64
+	NumOfStarts      uint64
+	NumOfEnds        uint64
+	NumOfDrops       uint64
+}
+
+func (e *Event) Clone() *Event {
+	toret := *e
+	toret.RuleID = slices.Clone(e.RuleID)
+	toret.SourceResourceID = slices.Clone(e.SourceResourceID)
+	toret.DestResourceID = slices.Clone(e.DestResourceID)
+	return &toret
 }
 
 type FlowConfig struct {
@@ -114,13 +128,15 @@ type FlowManager interface {
 	GetLogger() FlowLogger
 }
 
+type FlowEventAggregator interface {
+	ResetAggregationWindow() FlowEventAggregator
+	GetAggregatedEvents() []*Event
+}
+
 type FlowLogger interface {
+	ResetAggregationWindow() FlowEventAggregator
 	// StoreEvent stores a flow event
 	StoreEvent(flowEvent EventFields)
-	// GetEvents returns all stored events
-	GetEvents() []*Event
-	// DeleteEvents deletes events from the store
-	DeleteEvents([]uuid.UUID)
 	// Close closes the logger
 	Close()
 	// Enable enables the flow logger receiver
@@ -140,6 +156,11 @@ type Store interface {
 	Close()
 }
 
+type AggregatingStore interface {
+	FlowEventAggregator
+	Store
+}
+
 // ConnTracker defines the interface for connection tracking functionality
 type ConnTracker interface {
 	// Start begins tracking connections by listening for conntrack events.

client/internal/peer/conn_status.go

@@ -1,18 +1,5 @@
 package peer
 
-import (
-	log "github.com/sirupsen/logrus"
-)
-
-const (
-	// StatusIdle indicate the peer is in disconnected state
-	StatusIdle ConnStatus = iota
-	// StatusConnecting indicate the peer is in connecting state
-	StatusConnecting
-	// StatusConnected indicate the peer is in connected state
-	StatusConnected
-)
-
 // connStatusInputs is the primitive-valued snapshot of the state that drives the
 // tri-state connection classification. Extracted so the decision logic can be unit-tested
 // without constructing full Worker/Handshaker objects.
@@ -21,24 +8,7 @@ type connStatusInputs struct {
 	peerUsesRelay       bool // remote peer advertises relay support AND local has relay
 	relayConnected      bool // statusRelay reports Connected (independent of whether peer uses relay)
 	remoteSupportsICE   bool // remote peer sent ICE credentials
-	iceWorkerCreated    bool // local WorkerICE exists (false in force-relay mode)
+	iceWorkerCreated    bool // local ICE worker exists (false in force-relay mode)
 	iceStatusConnecting bool // statusICE is anything other than Disconnected
 	iceInProgress       bool // a negotiation is currently in flight
 }
-
-// ConnStatus describe the status of a peer's connection
-type ConnStatus int32
-
-func (s ConnStatus) String() string {
-	switch s {
-	case StatusConnecting:
-		return "Connecting"
-	case StatusConnected:
-		return "Connected"
-	case StatusIdle:
-		return "Idle"
-	default:
-		log.Errorf("unknown status: %d", s)
-		return "INVALID_PEER_CONNECTION_STATUS"
-	}
-}

client/internal/peer/conn_test.go

@@ -3,27 +3,33 @@ package peer
 import (
 	"context"
 	"fmt"
+	"net/netip"
 	"os"
 	"testing"
 	"time"
 
+	log "github.com/sirupsen/logrus"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 
 	"github.com/netbirdio/netbird/client/iface"
-	"github.com/netbirdio/netbird/client/internal/peer/dispatcher"
 	"github.com/netbirdio/netbird/client/internal/peer/guard"
 	"github.com/netbirdio/netbird/client/internal/peer/ice"
+	"github.com/netbirdio/netbird/client/internal/peer/metricsstages"
+	"github.com/netbirdio/netbird/client/internal/peer/signaling"
+	"github.com/netbirdio/netbird/client/internal/peer/status"
 	"github.com/netbirdio/netbird/client/internal/stdnet"
 	"github.com/netbirdio/netbird/util"
 )
 
-var testDispatcher = dispatcher.NewConnectionDispatcher()
-
 var connConf = ConnConfig{
 	Key:         "LLHf3Ma6z6mdLbriAJbqhX7+nM/B71lgw2+91q3LfhU=",
 	LocalKey:    "RRHf3Ma6z6mdLbriAJbqhX7+nM/B71lgw2+91q3LfhU=",
 	Timeout:     time.Second,
 	LocalWgPort: 51820,
+	WgConfig: WgConfig{
+		AllowedIps: []netip.Prefix{netip.MustParsePrefix("100.64.0.1/32")},
+	},
 	ICEConfig: ice.Config{
 		InterfaceBlackList: nil,
 	},
@@ -51,92 +57,37 @@ func TestConn_GetKey(t *testing.T) {
 	swWatcher := guard.NewSRWatcher(nil, nil, nil, connConf.ICEConfig)
 
 	sd := ServiceDependencies{
-		SrWatcher:          swWatcher,
-		PeerConnDispatcher: testDispatcher,
+		SrWatcher: swWatcher,
 	}
 	conn, err := NewConn(connConf, sd)
-	if err != nil {
-		return
-	}
+	require.NoError(t, err)
 
 	got := conn.GetKey()
 
 	assert.Equal(t, got, connConf.Key, "they should be equal")
 }
 
-func TestConn_OnRemoteOffer(t *testing.T) {
+// TestConn_DiscardMessagesWhenNotOpened: signal messages posted to a not yet
+// opened connection must be discarded without blocking or panicking.
+func TestConn_DiscardMessagesWhenNotOpened(t *testing.T) {
 	swWatcher := guard.NewSRWatcher(nil, nil, nil, connConf.ICEConfig)
 	sd := ServiceDependencies{
-		StatusRecorder:     NewRecorder("https://mgm"),
-		SrWatcher:          swWatcher,
-		PeerConnDispatcher: testDispatcher,
+		StatusRecorder: status.NewRecorder("https://mgm"),
+		SrWatcher:      swWatcher,
 	}
 	conn, err := NewConn(connConf, sd)
-	if err != nil {
-		return
-	}
+	require.NoError(t, err)
 
-	onNewOfferChan := make(chan struct{})
-
-	conn.handshaker.AddRelayListener(func(remoteOfferAnswer *OfferAnswer) {
-		onNewOfferChan <- struct{}{}
-	})
-
-	conn.OnRemoteOffer(OfferAnswer{
-		IceCredentials: IceCredentials{
+	offerAnswer := signaling.OfferAnswer{
+		IceCredentials: signaling.IceCredentials{
 			UFrag: "test",
 			Pwd:   "test",
 		},
-		WgListenPort: 0,
-		Version:      "",
-	})
-
-	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
-	defer cancel()
-
-	select {
-	case <-onNewOfferChan:
-		// success
-	case <-ctx.Done():
-		t.Error("expected to receive a new offer notification, but timed out")
-	}
-}
-
-func TestConn_OnRemoteAnswer(t *testing.T) {
-	swWatcher := guard.NewSRWatcher(nil, nil, nil, connConf.ICEConfig)
-	sd := ServiceDependencies{
-		StatusRecorder:     NewRecorder("https://mgm"),
-		SrWatcher:          swWatcher,
-		PeerConnDispatcher: testDispatcher,
-	}
-	conn, err := NewConn(connConf, sd)
-	if err != nil {
-		return
-	}
-
-	onNewOfferChan := make(chan struct{})
-
-	conn.handshaker.AddRelayListener(func(remoteOfferAnswer *OfferAnswer) {
-		onNewOfferChan <- struct{}{}
-	})
-
-	conn.OnRemoteAnswer(OfferAnswer{
-		IceCredentials: IceCredentials{
-			UFrag: "test",
-			Pwd:   "test",
-		},
-		WgListenPort: 0,
-		Version:      "",
-	})
-	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
-	defer cancel()
-
-	select {
-	case <-onNewOfferChan:
-		// success
-	case <-ctx.Done():
-		t.Error("expected to receive a new offer notification, but timed out")
 	}
+	conn.OnRemoteOffer(offerAnswer)
+	conn.OnRemoteAnswer(offerAnswer)
+	conn.OnRemoteCandidate(nil, nil)
+	conn.Close(false)
 }
 
 func TestConn_presharedKey(t *testing.T) {
@@ -304,3 +255,84 @@ func TestConn_presharedKey_RosenpassManaged(t *testing.T) {
 		t.Fatalf("expected non-nil presharedKey before Rosenpass manages PSK")
 	}
 }
+
+func newWGTimeoutTestConn(rosenpassEnabled bool, disconnected *[]string) *Conn {
+	cfg := ConnConfig{
+		Key:      "LLHf3Ma6z6mdLbriAJbqhX7+nM/B71lgw2+91q3LfhU=",
+		LocalKey: "RRHf3Ma6z6mdLbriAJbqhX7+nM/B71lgw2+91q3LfhU=",
+		WgConfig: WgConfig{RemoteKey: "LLHf3Ma6z6mdLbriAJbqhX7+nM/B71lgw2+91q3LfhU="},
+	}
+	if rosenpassEnabled {
+		cfg.RosenpassConfig = RosenpassConfig{PubKey: []byte("dummykey")}
+	}
+
+	conn := &Conn{
+		ctx:           context.Background(),
+		config:        cfg,
+		Log:           log.WithField("peer", cfg.Key),
+		metricsStages: &metricsstages.MetricsStages{},
+	}
+	conn.SetOnDisconnected(func(remotePeer string) {
+		*disconnected = append(*disconnected, remotePeer)
+	})
+	return conn
+}
+
+// TestConn_onWGDisconnected_EscalatesToRosenpassReset: repeated handshake
+// timeouts with rosenpass enabled mean the preshared keys have desynced. The
+// renewal exchange runs over the dead tunnel and cannot resync them, so after
+// wgTimeoutEscalationThreshold consecutive timeouts the conn must report the
+// peer disconnected, dropping its rosenpass state so the next configuration
+// programs the rendezvous key.
+func TestConn_onWGDisconnected_EscalatesToRosenpassReset(t *testing.T) {
+	var disconnected []string
+	conn := newWGTimeoutTestConn(true, &disconnected)
+
+	for i := 0; i < wgTimeoutEscalationThreshold-1; i++ {
+		conn.handleWGTimeout()
+	}
+	assert.Empty(t, disconnected, "escalation must not fire below the threshold")
+
+	conn.handleWGTimeout()
+	assert.Equal(t, []string{conn.config.WgConfig.RemoteKey}, disconnected,
+		"reaching the threshold must report the peer disconnected once")
+
+	for i := 0; i < wgTimeoutEscalationThreshold-1; i++ {
+		conn.handleWGTimeout()
+	}
+	assert.Len(t, disconnected, 1, "escalation must restart counting after firing")
+
+	conn.handleWGTimeout()
+	assert.Len(t, disconnected, 2, "continued timeouts must escalate again")
+}
+
+// TestConn_onWGDisconnected_CheckSuccessResetsEscalation: a successful
+// handshake between timeouts means the tunnel recovered; the counter must
+// start over.
+func TestConn_onWGDisconnected_CheckSuccessResetsEscalation(t *testing.T) {
+	var disconnected []string
+	conn := newWGTimeoutTestConn(true, &disconnected)
+
+	for i := 0; i < wgTimeoutEscalationThreshold-1; i++ {
+		conn.handleWGTimeout()
+	}
+	conn.handleWGCheckSuccess()
+
+	for i := 0; i < wgTimeoutEscalationThreshold-1; i++ {
+		conn.handleWGTimeout()
+	}
+	assert.Empty(t, disconnected, "handshake success must reset the timeout count")
+}
+
+// TestConn_onWGDisconnected_NoEscalationWithoutRosenpass: without rosenpass
+// there is no per-peer key state to reset; repeated timeouts must not report
+// disconnects.
+func TestConn_onWGDisconnected_NoEscalationWithoutRosenpass(t *testing.T) {
+	var disconnected []string
+	conn := newWGTimeoutTestConn(false, &disconnected)
+
+	for i := 0; i < wgTimeoutEscalationThreshold*3; i++ {
+		conn.handleWGTimeout()
+	}
+	assert.Empty(t, disconnected, "escalation must be limited to rosenpass connections")
+}

client/internal/peer/dispatcher/dispatcher.go

@@ -1,52 +0,0 @@
-package dispatcher
-
-import (
-	"sync"
-
-	"github.com/netbirdio/netbird/client/internal/peer/id"
-)
-
-type ConnectionListener struct {
-	OnConnected    func(peerID id.ConnID)
-	OnDisconnected func(peerID id.ConnID)
-}
-
-type ConnectionDispatcher struct {
-	listeners map[*ConnectionListener]struct{}
-	mu        sync.Mutex
-}
-
-func NewConnectionDispatcher() *ConnectionDispatcher {
-	return &ConnectionDispatcher{
-		listeners: make(map[*ConnectionListener]struct{}),
-	}
-}
-
-func (e *ConnectionDispatcher) AddListener(listener *ConnectionListener) {
-	e.mu.Lock()
-	defer e.mu.Unlock()
-	e.listeners[listener] = struct{}{}
-}
-
-func (e *ConnectionDispatcher) RemoveListener(listener *ConnectionListener) {
-	e.mu.Lock()
-	defer e.mu.Unlock()
-
-	delete(e.listeners, listener)
-}
-
-func (e *ConnectionDispatcher) NotifyConnected(peerConnID id.ConnID) {
-	e.mu.Lock()
-	defer e.mu.Unlock()
-	for listener := range e.listeners {
-		listener.OnConnected(peerConnID)
-	}
-}
-
-func (e *ConnectionDispatcher) NotifyDisconnected(peerConnID id.ConnID) {
-	e.mu.Lock()
-	defer e.mu.Unlock()
-	for listener := range e.listeners {
-		listener.OnDisconnected(peerConnID)
-	}
-}

client/internal/peer/event.go

@@ -0,0 +1,69 @@
+package peer
+
+import (
+	"time"
+
+	"github.com/pion/ice/v4"
+
+	"github.com/netbirdio/netbird/client/internal/peer/signaling"
+	"github.com/netbirdio/netbird/client/internal/peer/worker"
+	"github.com/netbirdio/netbird/route"
+)
+
+// event is a message processed by the Conn event loop. All mutable Conn state
+// is owned by that loop; producers deliver events through the mailbox and
+// never mutate Conn state directly.
+type event any
+
+// evClose asks the event loop to tear down the connection. done is closed
+// once the teardown finished.
+type evClose struct {
+	signalToRemote bool
+	done           chan struct{}
+}
+
+type evRemoteOffer struct {
+	offer signaling.OfferAnswer
+}
+
+type evRemoteAnswer struct {
+	answer signaling.OfferAnswer
+}
+
+type evRemoteCandidate struct {
+	candidate ice.Candidate
+	haRoutes  route.HAMap
+}
+
+type evICEReady struct {
+	priority worker.ConnPriority
+	info     worker.ICEConnInfo
+}
+
+type evICEDown struct {
+	sessionChanged bool
+}
+
+type evRelayReady struct {
+	info worker.RelayConnInfo
+}
+
+type evRelayDown struct{}
+
+// evRelayDialDone reports that the relay dial helper goroutine finished,
+// successfully or not, so the loop may dispatch a pending offer.
+type evRelayDialDone struct{}
+
+type evWGTimeout struct{}
+
+// evWGHandshake reports the first WireGuard handshake of the current watcher run.
+type evWGHandshake struct {
+	when time.Time
+}
+
+// evWGCheckOK reports a watcher check that observed a fresh handshake,
+// including handshakes of connections that were already up.
+type evWGCheckOK struct{}
+
+// evGuardTick asks the loop to send a new offer to restore connectivity.
+type evGuardTick struct{}

client/internal/peer/guard/guard.go

@@ -85,7 +85,11 @@ func (g *Guard) reconnectLoopWithRetry(ctx context.Context, callback func()) {
 	defer g.srWatcher.RemoveListener(srReconnectedChan)
 
 	ticker := g.initialTicker(ctx)
-	defer ticker.Stop()
+	defer func() {
+		// If backoff.Ticker.send is blocked, context.Done will not close the Ticker goroutine.
+		// We have to explicitly call Stop, even if we use backoff.WithContext.
+		ticker.Stop()
+	}()
 
 	tickerChannel := ticker.C
 

client/internal/peer/guard/guard_leak_test.go

@@ -0,0 +1,92 @@
+package guard
+
+import (
+	"context"
+	"runtime"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	log "github.com/sirupsen/logrus"
+
+	"github.com/netbirdio/netbird/client/internal/peer/ice"
+)
+
+func newTestGuard(status connStatusFunc) *Guard {
+	srw := NewSRWatcher(nil, nil, nil, ice.Config{})
+	return NewGuard(log.WithField("test", "guard"), status, 50*time.Millisecond, srw)
+}
+
+// countBackoffTickerGoroutines returns how many goroutines are currently sitting
+// in backoff/v4.(*Ticker).run (a ticker goroutine that has not exited).
+func countBackoffTickerGoroutines() int {
+	buf := make([]byte, 1<<25) // 32MB
+	n := runtime.Stack(buf, true)
+	return strings.Count(string(buf[:n]), "backoff/v4.(*Ticker).run")
+}
+
+// TestGuard_ReconnectTicker_NoGoroutineLeakOnShutdown reproduces a observed
+// leak: after a shutdown burst, ticker run/send goroutines stay parked
+// forever even though every reconnect loop has exited.
+func TestGuard_ReconnectTicker_NoGoroutineLeakOnShutdown(t *testing.T) {
+	before := countBackoffTickerGoroutines()
+
+	const peers = 6000
+	cancels := make([]context.CancelFunc, 0, peers)
+	var wg sync.WaitGroup
+
+	// A status check slower than the tick cadence. This models the real
+	// isConnectedOnAllWay/callback doing work: while the loop is busy in the
+	// handler, the ticker fires the next tick and parks in send(), because
+	// send() never selects on ctx.
+	slowStatus := func() ConnStatus {
+		time.Sleep(70 * time.Millisecond)
+		return ConnStatusConnected
+	}
+
+	for range peers {
+		g := newTestGuard(slowStatus)
+		ctx, cancel := context.WithCancel(context.Background())
+		cancels = append(cancels, cancel)
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			g.Start(ctx, func() {})
+		}()
+		// Force the live ticker to be a newReconnectTicker.
+		g.SetRelayedConnDisconnected()
+	}
+
+	// Let the replacement tickers get past their 800ms initial interval, so
+	// many are parked in send() waiting on the (slow) consumer when we tear
+	// everything down.
+	time.Sleep(1500 * time.Millisecond)
+
+	// Shutdown burst: cancel every peer at once, like engine teardown.
+	for _, c := range cancels {
+		c()
+	}
+
+	// Every reconnect loop must return
+	waitCh := make(chan struct{})
+	go func() { wg.Wait(); close(waitCh) }()
+	select {
+	case <-waitCh:
+	case <-time.After(30 * time.Second):
+		t.Fatal("not all reconnect loops returned after ctx cancel")
+	}
+
+	// Give any correctly-stopped ticker goroutines time to unwind.
+	for range 50 {
+		runtime.Gosched()
+		time.Sleep(10 * time.Millisecond)
+	}
+
+	leaked := countBackoffTickerGoroutines() - before
+	t.Logf("backoff Ticker.run goroutines still parked after teardown of %d peers: %d", peers, leaked)
+	if leaked > 0 {
+		t.Errorf("LEAK: %d backoff ticker goroutines parked after all reconnect loops exited "+
+			"(defer ticker.Stop() stops the initial ticker, not the live replacement)", leaked)
+	}
+}

client/internal/peer/guard/ice_monitor.go

@@ -21,8 +21,6 @@ const (
 )
 
 type ICEMonitor struct {
-	ReconnectCh chan struct{}
-
 	iFaceDiscover stdnet.ExternalIFaceDiscover
 	iceConfig     icemaker.Config
 	tickerPeriod  time.Duration
@@ -34,7 +32,6 @@ type ICEMonitor struct {
 func NewICEMonitor(iFaceDiscover stdnet.ExternalIFaceDiscover, config icemaker.Config, period time.Duration) *ICEMonitor {
 	log.Debugf("prepare ICE monitor with period: %s", period)
 	cm := &ICEMonitor{
-		ReconnectCh:   make(chan struct{}, 1),
 		iFaceDiscover: iFaceDiscover,
 		iceConfig:     config,
 		tickerPeriod:  period,

client/internal/peer/handshaker.go

@@ -1,246 +0,0 @@
-package peer
-
-import (
-	"context"
-	"errors"
-	"net/netip"
-	"sync"
-	"sync/atomic"
-
-	log "github.com/sirupsen/logrus"
-
-	"github.com/netbirdio/netbird/version"
-)
-
-var (
-	ErrSignalIsNotReady = errors.New("signal is not ready")
-)
-
-// IceCredentials ICE protocol credentials struct
-type IceCredentials struct {
-	UFrag string
-	Pwd   string
-}
-
-// OfferAnswer represents a session establishment offer or answer
-type OfferAnswer struct {
-	IceCredentials IceCredentials
-	// WgListenPort is a remote WireGuard listen port.
-	// This field is used when establishing a direct WireGuard connection without any proxy.
-	// We can set the remote peer's endpoint with this port.
-	WgListenPort int
-
-	// Version of NetBird Agent
-	Version string
-	// RosenpassPubKey is the Rosenpass public key of the remote peer when receiving this message
-	// This value is the local Rosenpass server public key when sending the message
-	RosenpassPubKey []byte
-	// RosenpassAddr is the Rosenpass server address (IP:port) of the remote peer when receiving this message
-	// This value is the local Rosenpass server address when sending the message
-	RosenpassAddr string
-
-	// relay server address
-	RelaySrvAddress string
-	// RelaySrvIP is the IP the remote peer is connected to on its
-	// relay server. Used as a dial target if DNS for RelaySrvAddress
-	// fails. Zero value if the peer did not advertise an IP.
-	RelaySrvIP netip.Addr
-	// SessionID is the unique identifier of the session, used to discard old messages
-	SessionID *ICESessionID
-}
-
-func (o *OfferAnswer) hasICECredentials() bool {
-	return o.IceCredentials.UFrag != "" && o.IceCredentials.Pwd != ""
-}
-
-type Handshaker struct {
-	mu            sync.Mutex
-	log           *log.Entry
-	config        ConnConfig
-	signaler      *Signaler
-	ice           *WorkerICE
-	relay         *WorkerRelay
-	metricsStages *MetricsStages
-	// relayListener is not blocking because the listener is using a goroutine to process the messages
-	// and it will only keep the latest message if multiple offers are received in a short time
-	// this is to avoid blocking the handshaker if the listener is doing some heavy processing
-	// and also to avoid processing old offers if multiple offers are received in a short time
-	// the listener will always process the latest offer
-	relayListener *AsyncOfferListener
-	iceListener   func(remoteOfferAnswer *OfferAnswer)
-
-	// remoteICESupported tracks whether the remote peer includes ICE credentials in its offers/answers.
-	// When false, the local side skips ICE listener dispatch and suppresses ICE credentials in responses.
-	remoteICESupported atomic.Bool
-
-	// remoteOffersCh is a channel used to wait for remote credentials to proceed with the connection
-	remoteOffersCh chan OfferAnswer
-	// remoteAnswerCh is a channel used to wait for remote credentials answer (confirmation of our offer) to proceed with the connection
-	remoteAnswerCh chan OfferAnswer
-}
-
-func NewHandshaker(log *log.Entry, config ConnConfig, signaler *Signaler, ice *WorkerICE, relay *WorkerRelay, metricsStages *MetricsStages) *Handshaker {
-	h := &Handshaker{
-		log:            log,
-		config:         config,
-		signaler:       signaler,
-		ice:            ice,
-		relay:          relay,
-		metricsStages:  metricsStages,
-		remoteOffersCh: make(chan OfferAnswer),
-		remoteAnswerCh: make(chan OfferAnswer),
-	}
-	// assume remote supports ICE until we learn otherwise from received offers
-	h.remoteICESupported.Store(ice != nil)
-	return h
-}
-
-func (h *Handshaker) RemoteICESupported() bool {
-	return h.remoteICESupported.Load()
-}
-
-func (h *Handshaker) AddRelayListener(offer func(remoteOfferAnswer *OfferAnswer)) {
-	h.relayListener = NewAsyncOfferListener(offer)
-}
-
-func (h *Handshaker) AddICEListener(offer func(remoteOfferAnswer *OfferAnswer)) {
-	h.iceListener = offer
-}
-
-func (h *Handshaker) Listen(ctx context.Context) {
-	for {
-		select {
-		case remoteOfferAnswer := <-h.remoteOffersCh:
-			h.log.Infof("received offer, running version %s, remote WireGuard listen port %d, session id: %s, remote ICE supported: %t", remoteOfferAnswer.Version, remoteOfferAnswer.WgListenPort, remoteOfferAnswer.SessionIDString(), remoteOfferAnswer.hasICECredentials())
-
-			// Record signaling received for reconnection attempts
-			if h.metricsStages != nil {
-				h.metricsStages.RecordSignalingReceived()
-			}
-
-			h.updateRemoteICEState(&remoteOfferAnswer)
-
-			if h.relayListener != nil {
-				h.relayListener.Notify(&remoteOfferAnswer)
-			}
-
-			if h.iceListener != nil && h.RemoteICESupported() {
-				h.iceListener(&remoteOfferAnswer)
-			}
-
-			if err := h.sendAnswer(); err != nil {
-				h.log.Errorf("failed to send remote offer confirmation: %s", err)
-				continue
-			}
-		case remoteOfferAnswer := <-h.remoteAnswerCh:
-			h.log.Infof("received answer, running version %s, remote WireGuard listen port %d, session id: %s, remote ICE supported: %t", remoteOfferAnswer.Version, remoteOfferAnswer.WgListenPort, remoteOfferAnswer.SessionIDString(), remoteOfferAnswer.hasICECredentials())
-
-			// Record signaling received for reconnection attempts
-			if h.metricsStages != nil {
-				h.metricsStages.RecordSignalingReceived()
-			}
-
-			h.updateRemoteICEState(&remoteOfferAnswer)
-
-			if h.relayListener != nil {
-				h.relayListener.Notify(&remoteOfferAnswer)
-			}
-
-			if h.iceListener != nil && h.RemoteICESupported() {
-				h.iceListener(&remoteOfferAnswer)
-			}
-		case <-ctx.Done():
-			h.log.Infof("stop listening for remote offers and answers")
-			return
-		}
-	}
-}
-
-func (h *Handshaker) SendOffer() error {
-	h.mu.Lock()
-	defer h.mu.Unlock()
-	return h.sendOffer()
-}
-
-// OnRemoteOffer handles an offer from the remote peer and returns true if the message was accepted, false otherwise
-// doesn't block, discards the message if connection wasn't ready
-func (h *Handshaker) OnRemoteOffer(offer OfferAnswer) {
-	select {
-	case h.remoteOffersCh <- offer:
-		return
-	default:
-		h.log.Warnf("skipping remote offer message because receiver not ready")
-		// connection might not be ready yet to receive so we ignore the message
-		return
-	}
-}
-
-// OnRemoteAnswer handles an offer from the remote peer and returns true if the message was accepted, false otherwise
-// doesn't block, discards the message if connection wasn't ready
-func (h *Handshaker) OnRemoteAnswer(answer OfferAnswer) {
-	select {
-	case h.remoteAnswerCh <- answer:
-		return
-	default:
-		// connection might not be ready yet to receive so we ignore the message
-		h.log.Warnf("skipping remote answer message because receiver not ready")
-		return
-	}
-}
-
-// sendOffer prepares local user credentials and signals them to the remote peer
-func (h *Handshaker) sendOffer() error {
-	if !h.signaler.Ready() {
-		return ErrSignalIsNotReady
-	}
-
-	offer := h.buildOfferAnswer()
-	h.log.Debugf("sending offer with serial: %s", offer.SessionIDString())
-
-	return h.signaler.SignalOffer(offer, h.config.Key)
-}
-
-func (h *Handshaker) sendAnswer() error {
-	answer := h.buildOfferAnswer()
-	h.log.Debugf("sending answer with serial: %s", answer.SessionIDString())
-
-	return h.signaler.SignalAnswer(answer, h.config.Key)
-}
-
-func (h *Handshaker) buildOfferAnswer() OfferAnswer {
-	answer := OfferAnswer{
-		WgListenPort:    h.config.LocalWgPort,
-		Version:         version.NetbirdVersion(),
-		RosenpassPubKey: h.config.RosenpassConfig.PubKey,
-		RosenpassAddr:   h.config.RosenpassConfig.Addr,
-	}
-
-	if h.ice != nil && h.RemoteICESupported() {
-		uFrag, pwd := h.ice.GetLocalUserCredentials()
-		sid := h.ice.SessionID()
-		answer.IceCredentials = IceCredentials{uFrag, pwd}
-		answer.SessionID = &sid
-	}
-
-	if addr, ip, err := h.relay.RelayInstanceAddress(); err == nil {
-		answer.RelaySrvAddress = addr
-		answer.RelaySrvIP = ip
-	}
-
-	return answer
-}
-
-func (h *Handshaker) updateRemoteICEState(offer *OfferAnswer) {
-	hasICE := offer.hasICECredentials()
-	prev := h.remoteICESupported.Swap(hasICE)
-	if prev != hasICE {
-		if hasICE {
-			h.log.Infof("remote peer started sending ICE credentials")
-		} else {
-			h.log.Infof("remote peer stopped sending ICE credentials")
-			if h.ice != nil {
-				h.ice.Close()
-			}
-		}
-	}
-}

client/internal/peer/handshaker_listener.go

@@ -1,62 +0,0 @@
-package peer
-
-import (
-	"sync"
-)
-
-type callbackFunc func(remoteOfferAnswer *OfferAnswer)
-
-func (oa *OfferAnswer) SessionIDString() string {
-	if oa.SessionID == nil {
-		return "unknown"
-	}
-	return oa.SessionID.String()
-}
-
-type AsyncOfferListener struct {
-	fn      callbackFunc
-	running bool
-	latest  *OfferAnswer
-	mu      sync.Mutex
-}
-
-func NewAsyncOfferListener(fn callbackFunc) *AsyncOfferListener {
-	return &AsyncOfferListener{
-		fn: fn,
-	}
-}
-
-func (o *AsyncOfferListener) Notify(remoteOfferAnswer *OfferAnswer) {
-	o.mu.Lock()
-	defer o.mu.Unlock()
-
-	// Store the latest offer
-	o.latest = remoteOfferAnswer
-
-	// If already running, the running goroutine will pick up this latest value
-	if o.running {
-		return
-	}
-
-	// Start processing
-	o.running = true
-
-	// Process in a goroutine to avoid blocking the caller
-	go func(remoteOfferAnswer *OfferAnswer) {
-		for {
-			o.fn(remoteOfferAnswer)
-
-			o.mu.Lock()
-			if o.latest == nil {
-				// No more work to do
-				o.running = false
-				o.mu.Unlock()
-				return
-			}
-			remoteOfferAnswer = o.latest
-			// Clear the latest to mark it as being processed
-			o.latest = nil
-			o.mu.Unlock()
-		}
-	}(remoteOfferAnswer)
-}

client/internal/peer/handshaker_listener_test.go

@@ -1,39 +0,0 @@
-package peer
-
-import (
-	"testing"
-	"time"
-)
-
-func Test_newOfferListener(t *testing.T) {
-	dummyOfferAnswer := &OfferAnswer{}
-	runChan := make(chan struct{}, 10)
-
-	longRunningFn := func(remoteOfferAnswer *OfferAnswer) {
-		time.Sleep(1 * time.Second)
-		runChan <- struct{}{}
-	}
-
-	hl := NewAsyncOfferListener(longRunningFn)
-
-	hl.Notify(dummyOfferAnswer)
-	hl.Notify(dummyOfferAnswer)
-	hl.Notify(dummyOfferAnswer)
-
-	// Wait for exactly 2 callbacks
-	for i := 0; i < 2; i++ {
-		select {
-		case <-runChan:
-		case <-time.After(3 * time.Second):
-			t.Fatal("Timeout waiting for callback")
-		}
-	}
-
-	// Verify no additional callbacks happen
-	select {
-	case <-runChan:
-		t.Fatal("Unexpected additional callback")
-	case <-time.After(100 * time.Millisecond):
-		t.Log("Correctly received exactly 2 callbacks")
-	}
-}

client/internal/peer/ice/session_id.go

@@ -1,4 +1,4 @@
-package peer
+package ice
 
 import (
 	"crypto/rand"
@@ -9,26 +9,26 @@ import (
 
 const sessionIDSize = 5
 
-type ICESessionID string
+type SessionID string
 
-// NewICESessionID generates a new session ID for distinguishing sessions
-func NewICESessionID() (ICESessionID, error) {
+// NewSessionID generates a new session ID for distinguishing sessions
+func NewSessionID() (SessionID, error) {
 	b := make([]byte, sessionIDSize)
 	if _, err := io.ReadFull(rand.Reader, b); err != nil {
 		return "", fmt.Errorf("failed to generate session ID: %w", err)
 	}
-	return ICESessionID(hex.EncodeToString(b)), nil
+	return SessionID(hex.EncodeToString(b)), nil
 }
 
-func ICESessionIDFromBytes(b []byte) (ICESessionID, error) {
+func SessionIDFromBytes(b []byte) (SessionID, error) {
 	if len(b) != sessionIDSize {
 		return "", fmt.Errorf("invalid session ID length: %d", len(b))
 	}
-	return ICESessionID(hex.EncodeToString(b)), nil
+	return SessionID(hex.EncodeToString(b)), nil
 }
 
 // Bytes returns the raw bytes of the session ID for protobuf serialization
-func (id ICESessionID) Bytes() ([]byte, error) {
+func (id SessionID) Bytes() ([]byte, error) {
 	if len(id) == 0 {
 		return nil, fmt.Errorf("ICE session ID is empty")
 	}
@@ -42,6 +42,6 @@ func (id ICESessionID) Bytes() ([]byte, error) {
 	return b, nil
 }
 
-func (id ICESessionID) String() string {
+func (id SessionID) String() string {
 	return string(id)
 }

client/internal/peer/iface.go

@@ -1,22 +0,0 @@
-package peer
-
-import (
-	"net"
-	"net/netip"
-	"time"
-
-	"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
-
-	"github.com/netbirdio/netbird/client/iface/configurer"
-	"github.com/netbirdio/netbird/client/iface/wgaddr"
-	"github.com/netbirdio/netbird/client/iface/wgproxy"
-)
-
-type WGIface interface {
-	UpdatePeer(peerKey string, allowedIps []netip.Prefix, keepAlive time.Duration, endpoint *net.UDPAddr, preSharedKey *wgtypes.Key) error
-	RemovePeer(peerKey string) error
-	GetStats() (map[string]configurer.WGStats, error)
-	GetProxy() wgproxy.Proxy
-	Address() wgaddr.Address
-	RemoveEndpointAddress(key string) error
-}

client/internal/peer/listener.go

@@ -1,11 +0,0 @@
-package peer
-
-// Listener is a callback type about the NetBird network connection state
-type Listener interface {
-	OnConnected()
-	OnDisconnected()
-	OnConnecting()
-	OnDisconnecting()
-	OnAddressChanged(string, string)
-	OnPeersListChanged(int)
-}

client/internal/peer/mailbox.go

@@ -0,0 +1,116 @@
+package peer
+
+import (
+	"sync"
+)
+
+// maxQueuedCandidates bounds the remote candidate queue; on overflow the
+// oldest candidate is dropped. Lost candidates are recovered by the next
+// offer exchange triggered by the guard.
+const maxQueuedCandidates = 128
+
+// mailbox is the coalescing inbox of the Conn event loop. Posting never
+// blocks. Per message kind either the latest value wins (offer, answer,
+// guard tick), the values queue in bounded FIFO order (candidates) or in
+// unbounded FIFO order (lifecycle and transport state changes, which are
+// low-volume and must not be lost). A new offer flushes the queued
+// candidates because they belong to the superseded session.
+type mailbox struct {
+	mu     sync.Mutex
+	closed bool
+
+	lifecycle  []event
+	transport  []event
+	offer      *evRemoteOffer
+	answer     *evRemoteAnswer
+	candidates []evRemoteCandidate
+	guardTick  bool
+
+	wake chan struct{}
+}
+
+func newMailbox() *mailbox {
+	return &mailbox{
+		wake: make(chan struct{}, 1),
+	}
+}
+
+// post stores the event and wakes the loop. It reports false if the mailbox
+// is already closed and the event was not accepted.
+func (m *mailbox) post(ev event) bool {
+	m.mu.Lock()
+	if m.closed {
+		m.mu.Unlock()
+		return false
+	}
+
+	switch e := ev.(type) {
+	case evClose:
+		m.lifecycle = append(m.lifecycle, e)
+	case evRemoteOffer:
+		m.offer = &e
+		m.candidates = nil
+	case evRemoteAnswer:
+		m.answer = &e
+	case evRemoteCandidate:
+		if len(m.candidates) >= maxQueuedCandidates {
+			m.candidates = m.candidates[1:]
+		}
+		m.candidates = append(m.candidates, e)
+	case evGuardTick:
+		m.guardTick = true
+	default:
+		m.transport = append(m.transport, ev)
+	}
+	m.mu.Unlock()
+
+	select {
+	case m.wake <- struct{}{}:
+	default:
+	}
+	return true
+}
+
+// drain returns the pending events in processing order: lifecycle first,
+// then transport state changes, the coalesced offer and answer, the queued
+// candidates and finally the guard tick.
+func (m *mailbox) drain() []event {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	return m.drainLocked()
+}
+
+// closeAndDrain marks the mailbox closed so further posts are rejected and
+// returns the events that were still pending.
+func (m *mailbox) closeAndDrain() []event {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.closed = true
+	return m.drainLocked()
+}
+
+func (m *mailbox) drainLocked() []event {
+	evs := make([]event, 0, len(m.lifecycle)+len(m.transport)+len(m.candidates)+3)
+	evs = append(evs, m.lifecycle...)
+	evs = append(evs, m.transport...)
+	if m.offer != nil {
+		evs = append(evs, *m.offer)
+	}
+	if m.answer != nil {
+		evs = append(evs, *m.answer)
+	}
+	for _, c := range m.candidates {
+		evs = append(evs, c)
+	}
+	if m.guardTick {
+		evs = append(evs, evGuardTick{})
+	}
+
+	m.lifecycle = nil
+	m.transport = nil
+	m.offer = nil
+	m.answer = nil
+	m.candidates = nil
+	m.guardTick = false
+	return evs
+}

client/internal/peer/mailbox_test.go

@@ -0,0 +1,128 @@
+package peer
+
+import (
+	"testing"
+
+	"github.com/netbirdio/netbird/client/internal/peer/signaling"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestMailbox_OfferCoalescing(t *testing.T) {
+	mb := newMailbox()
+
+	require.True(t, mb.post(evRemoteOffer{offer: signaling.OfferAnswer{WgListenPort: 1}}))
+	require.True(t, mb.post(evRemoteOffer{offer: signaling.OfferAnswer{WgListenPort: 2}}))
+	require.True(t, mb.post(evRemoteOffer{offer: signaling.OfferAnswer{WgListenPort: 3}}))
+
+	evs := mb.drain()
+	require.Len(t, evs, 1, "consecutive offers must coalesce to a single event")
+	offer, ok := evs[0].(evRemoteOffer)
+	require.True(t, ok, "coalesced event must be an offer")
+	assert.Equal(t, 3, offer.offer.WgListenPort, "the newest offer must win")
+}
+
+func TestMailbox_OfferFlushesCandidates(t *testing.T) {
+	mb := newMailbox()
+
+	require.True(t, mb.post(evRemoteCandidate{}))
+	require.True(t, mb.post(evRemoteCandidate{}))
+	require.True(t, mb.post(evRemoteOffer{offer: signaling.OfferAnswer{}}))
+
+	evs := mb.drain()
+	require.Len(t, evs, 1, "candidates of the superseded session must be flushed")
+	_, ok := evs[0].(evRemoteOffer)
+	assert.True(t, ok, "only the offer must remain after the flush")
+}
+
+func TestMailbox_CandidatesKeepOrderAfterOffer(t *testing.T) {
+	mb := newMailbox()
+
+	require.True(t, mb.post(evRemoteOffer{offer: signaling.OfferAnswer{}}))
+	require.True(t, mb.post(evRemoteCandidate{haRoutes: nil}))
+	require.True(t, mb.post(evRemoteCandidate{haRoutes: nil}))
+
+	evs := mb.drain()
+	require.Len(t, evs, 3)
+	_, ok := evs[0].(evRemoteOffer)
+	assert.True(t, ok, "offer must be processed before the candidates")
+	for _, ev := range evs[1:] {
+		_, ok := ev.(evRemoteCandidate)
+		assert.True(t, ok, "candidates posted after the offer must survive")
+	}
+}
+
+func TestMailbox_CandidateQueueBounded(t *testing.T) {
+	mb := newMailbox()
+
+	for i := 0; i < maxQueuedCandidates+10; i++ {
+		require.True(t, mb.post(evRemoteCandidate{}))
+	}
+
+	evs := mb.drain()
+	assert.Len(t, evs, maxQueuedCandidates, "candidate queue must stay bounded")
+}
+
+func TestMailbox_DrainOrder(t *testing.T) {
+	mb := newMailbox()
+
+	require.True(t, mb.post(evGuardTick{}))
+	require.True(t, mb.post(evRemoteAnswer{answer: signaling.OfferAnswer{}}))
+	require.True(t, mb.post(evRemoteOffer{offer: signaling.OfferAnswer{}}))
+	require.True(t, mb.post(evRelayDown{}))
+	require.True(t, mb.post(evICEDown{sessionChanged: true}))
+	require.True(t, mb.post(evClose{}))
+
+	evs := mb.drain()
+	require.Len(t, evs, 6)
+
+	_, ok := evs[0].(evClose)
+	assert.True(t, ok, "lifecycle events must come first")
+	_, ok = evs[1].(evRelayDown)
+	assert.True(t, ok, "transport events must keep FIFO order")
+	_, ok = evs[2].(evICEDown)
+	assert.True(t, ok, "transport events must keep FIFO order")
+	_, ok = evs[3].(evRemoteOffer)
+	assert.True(t, ok, "offer must come after transport events")
+	_, ok = evs[4].(evRemoteAnswer)
+	assert.True(t, ok, "answer must come after the offer")
+	_, ok = evs[5].(evGuardTick)
+	assert.True(t, ok, "guard tick must come last")
+}
+
+func TestMailbox_GuardTickCoalesced(t *testing.T) {
+	mb := newMailbox()
+
+	require.True(t, mb.post(evGuardTick{}))
+	require.True(t, mb.post(evGuardTick{}))
+	require.True(t, mb.post(evGuardTick{}))
+
+	evs := mb.drain()
+	assert.Len(t, evs, 1, "guard ticks must coalesce to a single event")
+}
+
+func TestMailbox_PostAfterCloseRejected(t *testing.T) {
+	mb := newMailbox()
+
+	require.True(t, mb.post(evRelayDown{}))
+	leftovers := mb.closeAndDrain()
+	assert.Len(t, leftovers, 1, "pending events must be returned on close")
+
+	assert.False(t, mb.post(evRelayDown{}), "posts must be rejected after close")
+	assert.Empty(t, mb.drain(), "no events must remain after close")
+}
+
+func TestMailbox_WakeSignal(t *testing.T) {
+	mb := newMailbox()
+
+	require.True(t, mb.post(evRelayDown{}))
+	require.True(t, mb.post(evGuardTick{}))
+
+	select {
+	case <-mb.wake:
+	default:
+		t.Fatal("wake signal must be pending after posts")
+	}
+
+	assert.Len(t, mb.drain(), 2, "a single wake must deliver all pending events")
+}

client/internal/peer/metricsstages/metrics_saver.go

@@ -1,4 +1,4 @@
-package peer
+package metricsstages
 
 import (
 	"sync"

client/internal/peer/metricsstages/metrics_saver_test.go

@@ -1,4 +1,4 @@
-package peer
+package metricsstages
 
 import (
 	"testing"

client/internal/peer/signaling/handshaker.go

@@ -0,0 +1,189 @@
+package signaling
+
+import (
+	"errors"
+	"net/netip"
+	"sync"
+	"sync/atomic"
+
+	log "github.com/sirupsen/logrus"
+
+	icemaker "github.com/netbirdio/netbird/client/internal/peer/ice"
+	relayClient "github.com/netbirdio/netbird/shared/relay/client"
+	"github.com/netbirdio/netbird/version"
+)
+
+var (
+	ErrSignalIsNotReady = errors.New("signal is not ready")
+)
+
+// IceCredentials ICE protocol credentials struct
+type IceCredentials struct {
+	UFrag string
+	Pwd   string
+}
+
+// OfferAnswer represents a session establishment offer or answer
+type OfferAnswer struct {
+	IceCredentials IceCredentials
+	// WgListenPort is a remote WireGuard listen port.
+	// This field is used when establishing a direct WireGuard connection without any proxy.
+	// We can set the remote peer's endpoint with this port.
+	WgListenPort int
+
+	// Version of NetBird Agent
+	Version string
+	// RosenpassPubKey is the Rosenpass public key of the remote peer when receiving this message
+	// This value is the local Rosenpass server public key when sending the message
+	RosenpassPubKey []byte
+	// RosenpassAddr is the Rosenpass server address (IP:port) of the remote peer when receiving this message
+	// This value is the local Rosenpass server address when sending the message
+	RosenpassAddr string
+
+	// relay server address
+	RelaySrvAddress string
+	// RelaySrvIP is the IP the remote peer is connected to on its
+	// relay server. Used as a dial target if DNS for RelaySrvAddress
+	// fails. Zero value if the peer did not advertise an IP.
+	RelaySrvIP netip.Addr
+	// SessionID is the unique identifier of the session, used to discard old messages
+	SessionID *icemaker.SessionID
+}
+
+func (o *OfferAnswer) HasICECredentials() bool {
+	return o.IceCredentials.UFrag != "" && o.IceCredentials.Pwd != ""
+}
+
+func (o *OfferAnswer) SessionIDString() string {
+	if o.SessionID == nil {
+		return "unknown"
+	}
+	return o.SessionID.String()
+}
+
+// Config carries the peer-specific values the Handshaker embeds into offers
+// and answers.
+type Config struct {
+	Key             string
+	LocalWgPort     int
+	RosenpassPubKey []byte
+	RosenpassAddr   string
+}
+
+// Credentials are the local ICE credentials and session id the Handshaker embeds in offers.
+type Credentials struct {
+	UFrag     string
+	Pwd       string
+	SessionID icemaker.SessionID
+}
+
+// ICEWorker is the subset of the ICE worker the Handshaker needs to build offers.
+type ICEWorker interface {
+	Credentials() Credentials
+	Close()
+}
+
+// Handshaker keeps the signaling protocol logic: building and sending offers
+// and answers and tracking whether the remote peer supports ICE. Incoming
+// message processing is driven by the Conn event loop.
+type Handshaker struct {
+	mu           sync.Mutex
+	log          *log.Entry
+	config       Config
+	signaler     *Signaler
+	ice          ICEWorker
+	relayManager *relayClient.Manager
+
+	// remoteICESupported tracks whether the remote peer includes ICE credentials in its offers/answers.
+	// When false, the local side skips ICE dispatch and suppresses ICE credentials in responses.
+	remoteICESupported atomic.Bool
+}
+
+func NewHandshaker(log *log.Entry, config Config, signaler *Signaler, ice ICEWorker, relayManager *relayClient.Manager) *Handshaker {
+	h := &Handshaker{
+		log:          log,
+		config:       config,
+		signaler:     signaler,
+		ice:          ice,
+		relayManager: relayManager,
+	}
+	// assume remote supports ICE until we learn otherwise from received offers
+	h.remoteICESupported.Store(ice != nil)
+	return h
+}
+
+func (h *Handshaker) RemoteICESupported() bool {
+	return h.remoteICESupported.Load()
+}
+
+func (h *Handshaker) SendOffer() error {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	return h.sendOffer()
+}
+
+func (h *Handshaker) SendAnswer() error {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	return h.sendAnswer()
+}
+
+// sendOffer prepares local user credentials and signals them to the remote peer
+func (h *Handshaker) sendOffer() error {
+	if !h.signaler.Ready() {
+		return ErrSignalIsNotReady
+	}
+
+	offer := h.buildOfferAnswer()
+	h.log.Debugf("sending offer with serial: %s", offer.SessionIDString())
+
+	return h.signaler.SignalOffer(offer, h.config.Key)
+}
+
+func (h *Handshaker) sendAnswer() error {
+	answer := h.buildOfferAnswer()
+	h.log.Debugf("sending answer with serial: %s", answer.SessionIDString())
+
+	return h.signaler.SignalAnswer(answer, h.config.Key)
+}
+
+func (h *Handshaker) buildOfferAnswer() OfferAnswer {
+	answer := OfferAnswer{
+		WgListenPort:    h.config.LocalWgPort,
+		Version:         version.NetbirdVersion(),
+		RosenpassPubKey: h.config.RosenpassPubKey,
+		RosenpassAddr:   h.config.RosenpassAddr,
+	}
+
+	if h.ice != nil && h.RemoteICESupported() {
+		creds := h.ice.Credentials()
+		answer.IceCredentials = IceCredentials{creds.UFrag, creds.Pwd}
+		sid := creds.SessionID
+		answer.SessionID = &sid
+	}
+
+	if addr, ip, err := h.relayManager.RelayInstanceAddress(); err == nil {
+		answer.RelaySrvAddress = addr
+		answer.RelaySrvIP = ip
+	}
+
+	return answer
+}
+
+// UpdateRemoteICEState refreshes the remote ICE support flag from a received
+// offer or answer and closes the ICE worker when the remote peer stopped
+// sending ICE credentials. Runs on the Conn event loop.
+func (h *Handshaker) UpdateRemoteICEState(offer *OfferAnswer) {
+	hasICE := offer.HasICECredentials()
+	prev := h.remoteICESupported.Swap(hasICE)
+	if prev != hasICE {
+		if hasICE {
+			h.log.Infof("remote peer started sending ICE credentials")
+		} else {
+			h.log.Infof("remote peer stopped sending ICE credentials")
+			if h.ice != nil {
+				h.ice.Close()
+			}
+		}
+	}
+}

client/internal/peer/signaling/signaler.go

@@ -1,4 +1,4 @@
-package peer
+package signaling
 
 import (
 	"github.com/pion/ice/v4"

client/internal/peer/state_dump/state_dump.go

@@ -1,4 +1,4 @@
-package peer
+package state_dump
 
 import (
 	"context"
@@ -6,11 +6,13 @@ import (
 	"time"
 
 	log "github.com/sirupsen/logrus"
+
+	"github.com/netbirdio/netbird/client/internal/peer/status"
 )
 
-type stateDump struct {
+type StateDump struct {
 	log    *log.Entry
-	status *Status
+	status *status.Recorder
 	key    string
 
 	sentOffer       int
@@ -26,15 +28,15 @@ type stateDump struct {
 	mu sync.Mutex
 }
 
-func newStateDump(key string, log *log.Entry, statusRecorder *Status) *stateDump {
-	return &stateDump{
+func NewStateDump(key string, log *log.Entry, statusRecorder *status.Recorder) *StateDump {
+	return &StateDump{
 		log:    log,
 		status: statusRecorder,
 		key:    key,
 	}
 }
 
-func (s *stateDump) Start(ctx context.Context) {
+func (s *StateDump) Start(ctx context.Context) {
 	ticker := time.NewTicker(10 * time.Minute)
 	defer ticker.Stop()
 
@@ -48,25 +50,25 @@ func (s *stateDump) Start(ctx context.Context) {
 	}
 }
 
-func (s *stateDump) RemoteOffer() {
+func (s *StateDump) RemoteOffer() {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 	s.remoteOffer++
 }
 
-func (s *stateDump) RemoteCandidate() {
+func (s *StateDump) RemoteCandidate() {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 	s.remoteCandidate++
 }
 
-func (s *stateDump) SendOffer() {
+func (s *StateDump) SendOffer() {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 	s.sentOffer++
 }
 
-func (s *stateDump) dumpState() {
+func (s *StateDump) dumpState() {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
@@ -80,41 +82,41 @@ func (s *stateDump) dumpState() {
 		status, s.sentOffer, s.remoteOffer, s.remoteAnswer, s.remoteCandidate, s.p2pConnected, s.switchToRelay, s.wgCheckSuccess, s.relayConnected, s.localProxies)
 }
 
-func (s *stateDump) RemoteAnswer() {
+func (s *StateDump) RemoteAnswer() {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 	s.remoteAnswer++
 }
 
-func (s *stateDump) P2PConnected() {
+func (s *StateDump) P2PConnected() {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
 	s.p2pConnected++
 }
 
-func (s *stateDump) SwitchToRelay() {
+func (s *StateDump) SwitchToRelay() {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
 	s.switchToRelay++
 }
 
-func (s *stateDump) WGcheckSuccess() {
+func (s *StateDump) WGcheckSuccess() {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
 	s.wgCheckSuccess++
 }
 
-func (s *stateDump) RelayConnected() {
+func (s *StateDump) RelayConnected() {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
 	s.relayConnected++
 }
 
-func (s *stateDump) NewLocalProxy() {
+func (s *StateDump) NewLocalProxy() {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 

client/internal/peer/status/conn_status.go

@@ -0,0 +1,31 @@
+package status
+
+import (
+	log "github.com/sirupsen/logrus"
+)
+
+const (
+	// StatusIdle indicate the peer is in disconnected state
+	StatusIdle ConnStatus = iota
+	// StatusConnecting indicate the peer is in connecting state
+	StatusConnecting
+	// StatusConnected indicate the peer is in connected state
+	StatusConnected
+)
+
+// ConnStatus describe the status of a peer's connection
+type ConnStatus int32
+
+func (s ConnStatus) String() string {
+	switch s {
+	case StatusConnecting:
+		return "Connecting"
+	case StatusConnected:
+		return "Connected"
+	case StatusIdle:
+		return "Idle"
+	default:
+		log.Errorf("unknown status: %d", s)
+		return "INVALID_PEER_CONNECTION_STATUS"
+	}
+}

client/internal/peer/status/conn_status_test.go

@@ -1,4 +1,4 @@
-package peer
+package status
 
 import (
 	"testing"

client/internal/peer/status/events.go

@@ -0,0 +1,48 @@
+package status
+
+import (
+	"slices"
+	"sync"
+
+	"github.com/netbirdio/netbird/client/proto"
+)
+
+type EventQueue struct {
+	maxSize int
+	events  []*proto.SystemEvent
+	mutex   sync.RWMutex
+}
+
+func NewEventQueue(size int) *EventQueue {
+	return &EventQueue{
+		maxSize: size,
+		events:  make([]*proto.SystemEvent, 0, size),
+	}
+}
+
+func (q *EventQueue) Add(event *proto.SystemEvent) {
+	q.mutex.Lock()
+	defer q.mutex.Unlock()
+
+	q.events = append(q.events, event)
+
+	if len(q.events) > q.maxSize {
+		q.events = q.events[len(q.events)-q.maxSize:]
+	}
+}
+
+func (q *EventQueue) GetAll() []*proto.SystemEvent {
+	q.mutex.RLock()
+	defer q.mutex.RUnlock()
+
+	return slices.Clone(q.events)
+}
+
+type EventSubscription struct {
+	id     string
+	events chan *proto.SystemEvent
+}
+
+func (s *EventSubscription) Events() <-chan *proto.SystemEvent {
+	return s.events
+}

client/internal/peer/status/full_status.go

@@ -0,0 +1,122 @@
+package status
+
+import (
+	"golang.org/x/exp/maps"
+	"google.golang.org/protobuf/types/known/durationpb"
+	"google.golang.org/protobuf/types/known/timestamppb"
+
+	"github.com/netbirdio/netbird/client/internal/relay"
+	"github.com/netbirdio/netbird/client/proto"
+)
+
+// FullStatus contains the full state held by the Recorder instance
+type FullStatus struct {
+	Peers                 []State
+	ManagementState       ManagementState
+	SignalState           SignalState
+	LocalPeerState        LocalPeerState
+	RosenpassState        RosenpassState
+	Relays                []relay.ProbeResult
+	NSGroupStates         []NSGroupState
+	NumOfForwardingRules  int
+	LazyConnectionEnabled bool
+	Events                []*proto.SystemEvent
+}
+
+// ToProto converts FullStatus to proto.FullStatus.
+func (fs FullStatus) ToProto() *proto.FullStatus {
+	pbFullStatus := proto.FullStatus{
+		ManagementState: &proto.ManagementState{},
+		SignalState:     &proto.SignalState{},
+		LocalPeerState:  &proto.LocalPeerState{},
+		Peers:           []*proto.PeerState{},
+	}
+
+	pbFullStatus.ManagementState.URL = fs.ManagementState.URL
+	pbFullStatus.ManagementState.Connected = fs.ManagementState.Connected
+	if err := fs.ManagementState.Error; err != nil {
+		pbFullStatus.ManagementState.Error = err.Error()
+	}
+
+	pbFullStatus.SignalState.URL = fs.SignalState.URL
+	pbFullStatus.SignalState.Connected = fs.SignalState.Connected
+	if err := fs.SignalState.Error; err != nil {
+		pbFullStatus.SignalState.Error = err.Error()
+	}
+
+	pbFullStatus.LocalPeerState.IP = fs.LocalPeerState.IP
+	pbFullStatus.LocalPeerState.Ipv6 = fs.LocalPeerState.IPv6
+	pbFullStatus.LocalPeerState.PubKey = fs.LocalPeerState.PubKey
+	pbFullStatus.LocalPeerState.KernelInterface = fs.LocalPeerState.KernelInterface
+	pbFullStatus.LocalPeerState.Fqdn = fs.LocalPeerState.FQDN
+	pbFullStatus.LocalPeerState.WgPort = int32(fs.LocalPeerState.WgPort)
+	pbFullStatus.LocalPeerState.RosenpassPermissive = fs.RosenpassState.Permissive
+	pbFullStatus.LocalPeerState.RosenpassEnabled = fs.RosenpassState.Enabled
+	pbFullStatus.NumberOfForwardingRules = int32(fs.NumOfForwardingRules)
+	pbFullStatus.LazyConnectionEnabled = fs.LazyConnectionEnabled
+
+	pbFullStatus.LocalPeerState.Networks = maps.Keys(fs.LocalPeerState.Routes)
+
+	for _, peerState := range fs.Peers {
+		networks := maps.Keys(peerState.GetRoutes())
+
+		pbPeerState := &proto.PeerState{
+			IP:                         peerState.IP,
+			Ipv6:                       peerState.IPv6,
+			PubKey:                     peerState.PubKey,
+			ConnStatus:                 peerState.ConnStatus.String(),
+			ConnStatusUpdate:           timestamppb.New(peerState.ConnStatusUpdate),
+			Relayed:                    peerState.Relayed,
+			LocalIceCandidateType:      peerState.LocalIceCandidateType,
+			RemoteIceCandidateType:     peerState.RemoteIceCandidateType,
+			LocalIceCandidateEndpoint:  peerState.LocalIceCandidateEndpoint,
+			RemoteIceCandidateEndpoint: peerState.RemoteIceCandidateEndpoint,
+			RelayAddress:               peerState.RelayServerAddress,
+			Fqdn:                       peerState.FQDN,
+			LastWireguardHandshake:     timestamppb.New(peerState.LastWireguardHandshake),
+			BytesRx:                    peerState.BytesRx,
+			BytesTx:                    peerState.BytesTx,
+			RosenpassEnabled:           peerState.RosenpassEnabled,
+			Networks:                   networks,
+			Latency:                    durationpb.New(peerState.Latency),
+			SshHostKey:                 peerState.SSHHostKey,
+		}
+		pbFullStatus.Peers = append(pbFullStatus.Peers, pbPeerState)
+	}
+
+	for _, relayState := range fs.Relays {
+		pbRelayState := &proto.RelayState{
+			URI:       relayState.URI,
+			Available: relayState.Err == nil,
+			Transport: relayState.Transport,
+		}
+		if err := relayState.Err; err != nil {
+			pbRelayState.Error = err.Error()
+		}
+		pbFullStatus.Relays = append(pbFullStatus.Relays, pbRelayState)
+	}
+
+	for _, dnsState := range fs.NSGroupStates {
+		var err string
+		if dnsState.Error != nil {
+			err = dnsState.Error.Error()
+		}
+
+		var servers []string
+		for _, server := range dnsState.Servers {
+			servers = append(servers, server.String())
+		}
+
+		pbDnsState := &proto.NSGroupState{
+			Servers: servers,
+			Domains: dnsState.Domains,
+			Enabled: dnsState.Enabled,
+			Error:   err,
+		}
+		pbFullStatus.DnsServers = append(pbFullStatus.DnsServers, pbDnsState)
+	}
+
+	pbFullStatus.Events = fs.Events
+
+	return &pbFullStatus
+}

client/internal/peer/status/notifier.go

@@ -1,4 +1,4 @@
-package peer
+package status
 
 import (
 	"sync"
@@ -11,6 +11,16 @@ const (
 	stateDisconnecting
 )
 
+// Listener is a callback type about the NetBird network connection state
+type Listener interface {
+	OnConnected()
+	OnDisconnected()
+	OnConnecting()
+	OnDisconnecting()
+	OnAddressChanged(string, string)
+	OnPeersListChanged(int)
+}
+
 type notifier struct {
 	serverStateLock    sync.Mutex
 	listenersLock      sync.Mutex

client/internal/peer/status/notifier_test.go

@@ -1,4 +1,4 @@
-package peer
+package status
 
 import (
 	"sync"

client/internal/peer/status/peer_state.go

@@ -0,0 +1,63 @@
+package status
+
+import (
+	"sync"
+	"time"
+
+	"golang.org/x/exp/maps"
+)
+
+// State contains the latest state of a peer
+type State struct {
+	Mux                        *sync.RWMutex
+	IP                         string
+	IPv6                       string
+	PubKey                     string
+	FQDN                       string
+	ConnStatus                 ConnStatus
+	ConnStatusUpdate           time.Time
+	Relayed                    bool
+	LocalIceCandidateType      string
+	RemoteIceCandidateType     string
+	LocalIceCandidateEndpoint  string
+	RemoteIceCandidateEndpoint string
+	RelayServerAddress         string
+	LastWireguardHandshake     time.Time
+	BytesTx                    int64
+	BytesRx                    int64
+	Latency                    time.Duration
+	RosenpassEnabled           bool
+	SSHHostKey                 []byte
+	routes                     map[string]struct{}
+}
+
+// AddRoute add a single route to routes map
+func (s *State) AddRoute(network string) {
+	s.Mux.Lock()
+	defer s.Mux.Unlock()
+	if s.routes == nil {
+		s.routes = make(map[string]struct{})
+	}
+	s.routes[network] = struct{}{}
+}
+
+// SetRoutes set state routes
+func (s *State) SetRoutes(routes map[string]struct{}) {
+	s.Mux.Lock()
+	defer s.Mux.Unlock()
+	s.routes = routes
+}
+
+// DeleteRoute removes a route from the network amp
+func (s *State) DeleteRoute(network string) {
+	s.Mux.Lock()
+	defer s.Mux.Unlock()
+	delete(s.routes, network)
+}
+
+// GetRoutes return routes map
+func (s *State) GetRoutes() map[string]struct{} {
+	s.Mux.RLock()
+	defer s.Mux.RUnlock()
+	return maps.Clone(s.routes)
+}

client/internal/peer/status/recorder_test.go

@@ -1,4 +1,4 @@
-package peer
+package status
 
 import (
 	"context"

client/internal/peer/status/route.go

@@ -1,4 +1,4 @@
-package peer
+package status
 
 import (
 	"net/netip"

client/internal/peer/status_alias.go

@@ -0,0 +1,36 @@
+package peer
+
+import "github.com/netbirdio/netbird/client/internal/peer/status"
+
+// Transitional aliases re-exporting the peer status recorder from its own
+// package. Callers are being migrated to reference the status package
+// directly; these aliases will be removed once the migration completes.
+type (
+	Status                   = status.Recorder
+	State                    = status.State
+	ConnStatus               = status.ConnStatus
+	FullStatus               = status.FullStatus
+	RouterState              = status.RouterState
+	LocalPeerState           = status.LocalPeerState
+	SignalState              = status.SignalState
+	ManagementState          = status.ManagementState
+	RosenpassState           = status.RosenpassState
+	NSGroupState             = status.NSGroupState
+	ResolvedDomainInfo       = status.ResolvedDomainInfo
+	StatusChangeSubscription = status.StatusChangeSubscription
+	EventQueue               = status.EventQueue
+	EventSubscription        = status.EventSubscription
+	WGIfaceStatus            = status.WGIfaceStatus
+	Listener                 = status.Listener
+	EventListener            = status.EventListener
+)
+
+const (
+	StatusIdle       = status.StatusIdle
+	StatusConnecting = status.StatusConnecting
+	StatusConnected  = status.StatusConnected
+)
+
+var (
+	NewRecorder = status.NewRecorder
+)

client/internal/peer/wg_watcher/wg_watcher.go

@@ -1,4 +1,4 @@
-package peer
+package wg_watcher
 
 import (
 	"context"
@@ -9,6 +9,7 @@ import (
 	log "github.com/sirupsen/logrus"
 
 	"github.com/netbirdio/netbird/client/iface/configurer"
+	"github.com/netbirdio/netbird/client/internal/peer/state_dump"
 )
 
 const (
@@ -28,7 +29,7 @@ type WGWatcher struct {
 	log           *log.Entry
 	wgIfaceStater WGInterfaceStater
 	peerKey       string
-	stateDump     *stateDump
+	stateDump     *state_dump.StateDump
 
 	enabled   bool
 	muEnabled sync.Mutex
@@ -38,7 +39,7 @@ type WGWatcher struct {
 	resetCh chan struct{}
 }
 
-func NewWGWatcher(log *log.Entry, wgIfaceStater WGInterfaceStater, peerKey string, stateDump *stateDump) *WGWatcher {
+func NewWGWatcher(log *log.Entry, wgIfaceStater WGInterfaceStater, peerKey string, stateDump *state_dump.StateDump) *WGWatcher {
 	return &WGWatcher{
 		log:           log,
 		wgIfaceStater: wgIfaceStater,
@@ -71,9 +72,11 @@ func (w *WGWatcher) PrepareInitialHandshake() (ok bool) {
 
 // EnableWgWatcher runs the WireGuard watcher loop using the handshake baseline captured by
 // PrepareInitialHandshake. The watcher runs until ctx is cancelled. Caller is responsible
-// for context lifecycle management.
-func (w *WGWatcher) EnableWgWatcher(ctx context.Context, enabledTime time.Time, onDisconnectedFn func(), onHandshakeSuccessFn func(when time.Time)) {
-	w.periodicHandshakeCheck(ctx, onDisconnectedFn, onHandshakeSuccessFn, enabledTime, w.initialHandshake)
+// for context lifecycle management. onHandshakeSuccessFn is called only for the first
+// handshake observed by this run, onCheckSuccessFn for every check that observed a fresh
+// handshake, including the first.
+func (w *WGWatcher) EnableWgWatcher(ctx context.Context, enabledTime time.Time, onDisconnectedFn func(), onHandshakeSuccessFn func(when time.Time), onCheckSuccessFn func()) {
+	w.periodicHandshakeCheck(ctx, onDisconnectedFn, onHandshakeSuccessFn, onCheckSuccessFn, enabledTime, w.initialHandshake)
 
 	w.muEnabled.Lock()
 	w.enabled = false
@@ -90,7 +93,7 @@ func (w *WGWatcher) Reset() {
 }
 
 // wgStateCheck help to check the state of the WireGuard handshake and relay connection
-func (w *WGWatcher) periodicHandshakeCheck(ctx context.Context, onDisconnectedFn func(), onHandshakeSuccessFn func(when time.Time), enabledTime time.Time, initialHandshake time.Time) {
+func (w *WGWatcher) periodicHandshakeCheck(ctx context.Context, onDisconnectedFn func(), onHandshakeSuccessFn func(when time.Time), onCheckSuccessFn func(), enabledTime time.Time, initialHandshake time.Time) {
 	w.log.Infof("WireGuard watcher started")
 
 	timer := time.NewTimer(wgHandshakeOvertime)
@@ -117,6 +120,10 @@ func (w *WGWatcher) periodicHandshakeCheck(ctx context.Context, onDisconnectedFn
 				}
 			}
 
+			if onCheckSuccessFn != nil && ctx.Err() == nil {
+				onCheckSuccessFn()
+			}
+
 			lastHandshake = *handshake
 
 			resetTime := time.Until(handshake.Add(checkPeriod))

client/internal/peer/wg_watcher/wg_watcher_test.go

@@ -0,0 +1,179 @@
+package wg_watcher
+
+import (
+	"context"
+	"sync"
+	"testing"
+	"time"
+
+	log "github.com/sirupsen/logrus"
+	"github.com/stretchr/testify/require"
+
+	"github.com/netbirdio/netbird/client/iface/configurer"
+	"github.com/netbirdio/netbird/client/internal/peer/state_dump"
+	"github.com/netbirdio/netbird/client/internal/peer/status"
+)
+
+type MocWgIface struct {
+	stop bool
+}
+
+func (m *MocWgIface) GetStats() (map[string]configurer.WGStats, error) {
+	return map[string]configurer.WGStats{}, nil
+}
+
+func (m *MocWgIface) disconnect() {
+	m.stop = true
+}
+
+type mockHandshakeStats struct {
+	mu        sync.Mutex
+	handshake time.Time
+}
+
+func (m *mockHandshakeStats) GetStats() (map[string]configurer.WGStats, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	return map[string]configurer.WGStats{"": {LastHandshake: m.handshake}}, nil
+}
+
+func (m *mockHandshakeStats) advance() {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.handshake = time.Now()
+}
+
+// TestWGWatcher_CheckSuccessCallback: onCheckSuccessFn must fire for a fresh
+// handshake even when the watcher started with an existing handshake baseline,
+// the case where onHandshakeSuccessFn stays silent.
+func TestWGWatcher_CheckSuccessCallback(t *testing.T) {
+	// checkPeriod bounds how stale a handshake may be before the watcher treats it
+	// as a suspended-machine timeout. The first check fires after wgHandshakeOvertime,
+	// so keep checkPeriod well above any scheduling jitter to avoid a false timeout
+	// converting the expected success into a disconnect on a loaded runner.
+	checkPeriod = 1 * time.Minute
+	wgHandshakeOvertime = 1 * time.Second
+
+	mlog := log.WithField("peer", "tet")
+	// Use an old baseline so advance() yields a strictly newer handshake even on
+	// platforms with coarse clock resolution (Windows), where two time.Now() calls
+	// microseconds apart can return the same instant and read as a timed-out handshake.
+	stats := &mockHandshakeStats{handshake: time.Now().Add(-time.Hour)}
+	watcher := NewWGWatcher(mlog, stats, "", state_dump.NewStateDump("peer", mlog, &status.Recorder{}))
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	require.True(t, watcher.PrepareInitialHandshake())
+
+	firstHandshake := make(chan struct{}, 1)
+	checkSuccess := make(chan struct{}, 1)
+	watcherDone := make(chan struct{})
+	go func() {
+		defer close(watcherDone)
+		watcher.EnableWgWatcher(ctx, time.Now(), func() {}, func(when time.Time) {
+			firstHandshake <- struct{}{}
+		}, func() {
+			select {
+			case checkSuccess <- struct{}{}:
+			default:
+			}
+		})
+	}()
+
+	stats.advance()
+
+	select {
+	case <-checkSuccess:
+	case <-time.After(10 * time.Second):
+		t.Errorf("timeout waiting for check success callback")
+	}
+
+	select {
+	case <-firstHandshake:
+		t.Errorf("first-handshake callback must not fire for a non-zero baseline")
+	default:
+	}
+
+	// Wait for the watcher goroutine to exit so it cannot race with other
+	// tests mutating the package-level check timing variables.
+	cancel()
+	<-watcherDone
+}
+
+func TestWGWatcher_EnableWgWatcher(t *testing.T) {
+	checkPeriod = 5 * time.Second
+	wgHandshakeOvertime = 1 * time.Second
+
+	mlog := log.WithField("peer", "tet")
+	mocWgIface := &MocWgIface{}
+	watcher := NewWGWatcher(mlog, mocWgIface, "", state_dump.NewStateDump("peer", mlog, &status.Recorder{}))
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	ok := watcher.PrepareInitialHandshake()
+	require.True(t, ok, "watcher should not be enabled yet")
+
+	onDisconnected := make(chan struct{}, 1)
+	go watcher.EnableWgWatcher(ctx, time.Now(), func() {
+		mlog.Infof("onDisconnectedFn")
+		onDisconnected <- struct{}{}
+	}, func(when time.Time) {
+		mlog.Infof("onHandshakeSuccess: %v", when)
+	}, nil)
+
+	// wait for initial reading
+	time.Sleep(2 * time.Second)
+	mocWgIface.disconnect()
+
+	select {
+	case <-onDisconnected:
+	case <-time.After(10 * time.Second):
+		t.Errorf("timeout")
+	}
+}
+
+func TestWGWatcher_ReEnable(t *testing.T) {
+	checkPeriod = 5 * time.Second
+	wgHandshakeOvertime = 1 * time.Second
+
+	mlog := log.WithField("peer", "tet")
+	mocWgIface := &MocWgIface{}
+	watcher := NewWGWatcher(mlog, mocWgIface, "", state_dump.NewStateDump("peer", mlog, &status.Recorder{}))
+
+	ctx, cancel := context.WithCancel(context.Background())
+	ok := watcher.PrepareInitialHandshake()
+	require.True(t, ok, "watcher should not be enabled yet")
+
+	wg := &sync.WaitGroup{}
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		watcher.EnableWgWatcher(ctx, time.Now(), func() {}, func(when time.Time) {}, nil)
+	}()
+	cancel()
+
+	wg.Wait()
+
+	// Re-enable with a new context
+	ctx, cancel = context.WithCancel(context.Background())
+	defer cancel()
+
+	ok = watcher.PrepareInitialHandshake()
+	require.True(t, ok, "watcher should be re-enabled after the previous run stopped")
+
+	onDisconnected := make(chan struct{}, 1)
+	go watcher.EnableWgWatcher(ctx, time.Now(), func() {
+		onDisconnected <- struct{}{}
+	}, func(when time.Time) {}, nil)
+
+	time.Sleep(2 * time.Second)
+	mocWgIface.disconnect()
+
+	select {
+	case <-onDisconnected:
+	case <-time.After(10 * time.Second):
+		t.Errorf("timeout")
+	}
+}

client/internal/peer/wg_watcher_test.go

@@ -1,102 +0,0 @@
-package peer
-
-import (
-	"context"
-	"sync"
-	"testing"
-	"time"
-
-	log "github.com/sirupsen/logrus"
-	"github.com/stretchr/testify/require"
-
-	"github.com/netbirdio/netbird/client/iface/configurer"
-)
-
-type MocWgIface struct {
-	stop bool
-}
-
-func (m *MocWgIface) GetStats() (map[string]configurer.WGStats, error) {
-	return map[string]configurer.WGStats{}, nil
-}
-
-func (m *MocWgIface) disconnect() {
-	m.stop = true
-}
-
-func TestWGWatcher_EnableWgWatcher(t *testing.T) {
-	checkPeriod = 5 * time.Second
-	wgHandshakeOvertime = 1 * time.Second
-
-	mlog := log.WithField("peer", "tet")
-	mocWgIface := &MocWgIface{}
-	watcher := NewWGWatcher(mlog, mocWgIface, "", newStateDump("peer", mlog, &Status{}))
-
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
-
-	ok := watcher.PrepareInitialHandshake()
-	require.True(t, ok, "watcher should not be enabled yet")
-
-	onDisconnected := make(chan struct{}, 1)
-	go watcher.EnableWgWatcher(ctx, time.Now(), func() {
-		mlog.Infof("onDisconnectedFn")
-		onDisconnected <- struct{}{}
-	}, func(when time.Time) {
-		mlog.Infof("onHandshakeSuccess: %v", when)
-	})
-
-	// wait for initial reading
-	time.Sleep(2 * time.Second)
-	mocWgIface.disconnect()
-
-	select {
-	case <-onDisconnected:
-	case <-time.After(10 * time.Second):
-		t.Errorf("timeout")
-	}
-}
-
-func TestWGWatcher_ReEnable(t *testing.T) {
-	checkPeriod = 5 * time.Second
-	wgHandshakeOvertime = 1 * time.Second
-
-	mlog := log.WithField("peer", "tet")
-	mocWgIface := &MocWgIface{}
-	watcher := NewWGWatcher(mlog, mocWgIface, "", newStateDump("peer", mlog, &Status{}))
-
-	ctx, cancel := context.WithCancel(context.Background())
-	ok := watcher.PrepareInitialHandshake()
-	require.True(t, ok, "watcher should not be enabled yet")
-
-	wg := &sync.WaitGroup{}
-	wg.Add(1)
-	go func() {
-		defer wg.Done()
-		watcher.EnableWgWatcher(ctx, time.Now(), func() {}, func(when time.Time) {})
-	}()
-	cancel()
-
-	wg.Wait()
-
-	// Re-enable with a new context
-	ctx, cancel = context.WithCancel(context.Background())
-	defer cancel()
-
-	ok = watcher.PrepareInitialHandshake()
-	require.True(t, ok, "watcher should be re-enabled after the previous run stopped")
-
-	onDisconnected := make(chan struct{}, 1)
-	go watcher.EnableWgWatcher(ctx, time.Now(), func() {
-		onDisconnected <- struct{}{}
-	}, func(when time.Time) {})
-
-	time.Sleep(2 * time.Second)
-	mocWgIface.disconnect()
-
-	select {
-	case <-onDisconnected:
-	case <-time.After(10 * time.Second):
-		t.Errorf("timeout")
-	}
-}

client/internal/peer/worker/priority.go

@@ -1,4 +1,4 @@
-package conntype
+package worker
 
 import (
 	"fmt"

client/internal/peer/worker/worker_ice.go

@@ -1,4 +1,4 @@
-package peer
+package worker
 
 import (
 	"context"
@@ -13,8 +13,9 @@ import (
 
 	"github.com/netbirdio/netbird/client/iface"
 	"github.com/netbirdio/netbird/client/iface/udpmux"
-	"github.com/netbirdio/netbird/client/internal/peer/conntype"
 	icemaker "github.com/netbirdio/netbird/client/internal/peer/ice"
+	"github.com/netbirdio/netbird/client/internal/peer/signaling"
+	"github.com/netbirdio/netbird/client/internal/peer/status"
 	"github.com/netbirdio/netbird/client/internal/portforward"
 	"github.com/netbirdio/netbird/client/internal/stdnet"
 	"github.com/netbirdio/netbird/route"
@@ -32,57 +33,68 @@ type ICEConnInfo struct {
 	RelayedOnLocal             bool
 }
 
-type WorkerICE struct {
-	ctx               context.Context
-	log               *log.Entry
-	config            ConnConfig
-	conn              *Conn
-	signaler          *Signaler
-	iFaceDiscover     stdnet.ExternalIFaceDiscover
-	statusRecorder    *Status
-	hasRelayOnLocally bool
+type ICEDependencies struct {
+	Signaler           *signaling.Signaler
+	IFaceDiscover      stdnet.ExternalIFaceDiscover
+	StatusRecorder     *status.Recorder
+	PortForwardManager *portforward.Manager
+}
+
+type ICE struct {
+	log                *log.Entry
+	key                string
+	iceConfig          icemaker.Config
+	isController       bool
+	onConnReady        func(priority ConnPriority, iceConnInfo ICEConnInfo)
+	onStatusDisconnect func(sessionChanged bool)
+	signaler           *signaling.Signaler
+	iFaceDiscover      stdnet.ExternalIFaceDiscover
+	statusRecorder     *status.Recorder
+	portForwardManager *portforward.Manager
+	hasRelayOnLocally  bool
 
 	agent             *icemaker.ThreadSafeAgent
 	agentDialerCancel context.CancelFunc
 	agentConnecting   bool      // while it is true, drop all incoming offers
 	lastSuccess       time.Time // with this avoid the too frequent ICE agent recreation
+	// connectedAgent is the agent whose connection was last reported ready; guarded by muxAgent
+	connectedAgent *icemaker.ThreadSafeAgent
 	// remoteSessionID represents the peer's session identifier from the latest remote offer.
-	remoteSessionID ICESessionID
+	remoteSessionID icemaker.SessionID
 	// sessionID is used to track the current session ID of the ICE agent
 	// increase by one when disconnecting the agent
 	// with it the remote peer can discard the already deprecated offer/answer
 	// Without it the remote peer may recreate a workable ICE connection
-	sessionID            ICESessionID
+	sessionID            icemaker.SessionID
 	remoteSessionChanged bool
 	muxAgent             sync.Mutex
 
 	localUfrag string
 	localPwd   string
 
-	// we record the last known state of the ICE agent to avoid duplicate on disconnected events
-	lastKnownState ice.ConnectionState
-
 	// portForwardAttempted tracks if we've already tried port forwarding this session
 	portForwardAttempted bool
 }
 
-func NewWorkerICE(ctx context.Context, log *log.Entry, config ConnConfig, conn *Conn, signaler *Signaler, ifaceDiscover stdnet.ExternalIFaceDiscover, statusRecorder *Status, hasRelayOnLocally bool) (*WorkerICE, error) {
-	sessionID, err := NewICESessionID()
+func NewICE(log *log.Entry, key string, iceConfig icemaker.Config, isController bool, onConnReady func(ConnPriority, ICEConnInfo), onStatusDisconnect func(bool), services ICEDependencies, hasRelayOnLocally bool) (*ICE, error) {
+	sessionID, err := icemaker.NewSessionID()
 	if err != nil {
 		return nil, err
 	}
 
-	w := &WorkerICE{
-		ctx:               ctx,
-		log:               log,
-		config:            config,
-		conn:              conn,
-		signaler:          signaler,
-		iFaceDiscover:     ifaceDiscover,
-		statusRecorder:    statusRecorder,
-		hasRelayOnLocally: hasRelayOnLocally,
-		lastKnownState:    ice.ConnectionStateDisconnected,
-		sessionID:         sessionID,
+	w := &ICE{
+		log:                log,
+		key:                key,
+		iceConfig:          iceConfig,
+		isController:       isController,
+		onConnReady:        onConnReady,
+		onStatusDisconnect: onStatusDisconnect,
+		signaler:           services.Signaler,
+		iFaceDiscover:      services.IFaceDiscover,
+		statusRecorder:     services.StatusRecorder,
+		portForwardManager: services.PortForwardManager,
+		hasRelayOnLocally:  hasRelayOnLocally,
+		sessionID:          sessionID,
 	}
 
 	localUfrag, localPwd, err := icemaker.GenerateICECredentials()
@@ -94,7 +106,7 @@ func NewWorkerICE(ctx context.Context, log *log.Entry, config ConnConfig, conn *
 	return w, nil
 }
 
-func (w *WorkerICE) OnNewOffer(remoteOfferAnswer *OfferAnswer) {
+func (w *ICE) OnNewOffer(ctx context.Context, remoteOfferAnswer *signaling.OfferAnswer) {
 	w.log.Debugf("OnNewOffer for ICE, serial: %s", remoteOfferAnswer.SessionIDString())
 	w.muxAgent.Lock()
 	defer w.muxAgent.Unlock()
@@ -118,7 +130,7 @@ func (w *WorkerICE) OnNewOffer(remoteOfferAnswer *OfferAnswer) {
 			}
 		}
 
-		sessionID, err := NewICESessionID()
+		sessionID, err := icemaker.NewSessionID()
 		if err != nil {
 			w.log.Errorf("failed to create new session ID: %s", err)
 		}
@@ -136,8 +148,8 @@ func (w *WorkerICE) OnNewOffer(remoteOfferAnswer *OfferAnswer) {
 	if remoteOfferAnswer.SessionID != nil {
 		w.log.Debugf("recreate ICE agent: %s / %s", w.sessionID, *remoteOfferAnswer.SessionID)
 	}
-	dialerCtx, dialerCancel := context.WithCancel(w.ctx)
-	agent, err := w.reCreateAgent(dialerCancel, preferredCandidateTypes)
+	dialerCtx, dialerCancel := context.WithCancel(ctx)
+	agent, err := w.reCreateAgent(ctx, dialerCancel, preferredCandidateTypes)
 	if err != nil {
 		w.log.Errorf("failed to recreate ICE Agent: %s", err)
 		return
@@ -151,14 +163,14 @@ func (w *WorkerICE) OnNewOffer(remoteOfferAnswer *OfferAnswer) {
 		w.remoteSessionID = ""
 	}
 
-	go w.connect(dialerCtx, agent, remoteOfferAnswer)
+	go w.connect(dialerCtx, dialerCancel, agent, remoteOfferAnswer)
 }
 
 // OnRemoteCandidate Handles ICE connection Candidate provided by the remote peer.
-func (w *WorkerICE) OnRemoteCandidate(candidate ice.Candidate, haRoutes route.HAMap) {
+func (w *ICE) OnRemoteCandidate(candidate ice.Candidate, haRoutes route.HAMap) {
 	w.muxAgent.Lock()
 	defer w.muxAgent.Unlock()
-	w.log.Debugf("OnRemoteCandidate from peer %s -> %s", w.config.Key, candidate.String())
+	w.log.Debugf("OnRemoteCandidate from peer %s -> %s", w.key, candidate.String())
 	if w.agent == nil {
 		w.log.Warnf("ICE Agent is not initialized yet")
 		return
@@ -185,18 +197,24 @@ func (w *WorkerICE) OnRemoteCandidate(candidate ice.Candidate, haRoutes route.HA
 	}
 }
 
-func (w *WorkerICE) GetLocalUserCredentials() (frag string, pwd string) {
-	return w.localUfrag, w.localPwd
+func (w *ICE) Credentials() signaling.Credentials {
+	w.muxAgent.Lock()
+	defer w.muxAgent.Unlock()
+	return signaling.Credentials{
+		UFrag:     w.localUfrag,
+		Pwd:       w.localPwd,
+		SessionID: w.sessionID,
+	}
 }
 
-func (w *WorkerICE) InProgress() bool {
+func (w *ICE) InProgress() bool {
 	w.muxAgent.Lock()
 	defer w.muxAgent.Unlock()
 
 	return w.agentConnecting
 }
 
-func (w *WorkerICE) Close() {
+func (w *ICE) Close() {
 	w.muxAgent.Lock()
 	defer w.muxAgent.Unlock()
 
@@ -212,10 +230,10 @@ func (w *WorkerICE) Close() {
 	w.agent = nil
 }
 
-func (w *WorkerICE) reCreateAgent(dialerCancel context.CancelFunc, candidates []ice.CandidateType) (*icemaker.ThreadSafeAgent, error) {
+func (w *ICE) reCreateAgent(ctx context.Context, dialerCancel context.CancelFunc, candidates []ice.CandidateType) (*icemaker.ThreadSafeAgent, error) {
 	w.portForwardAttempted = false
 
-	agent, err := icemaker.NewAgent(w.ctx, w.iFaceDiscover, w.config.ICEConfig, candidates, w.localUfrag, w.localPwd)
+	agent, err := icemaker.NewAgent(ctx, w.iFaceDiscover, w.iceConfig, candidates, w.localUfrag, w.localPwd)
 	if err != nil {
 		return nil, fmt.Errorf("create agent: %w", err)
 	}
@@ -237,7 +255,7 @@ func (w *WorkerICE) reCreateAgent(dialerCancel context.CancelFunc, candidates []
 	return agent, nil
 }
 
-func (w *WorkerICE) SessionID() ICESessionID {
+func (w *ICE) getSessionID() icemaker.SessionID {
 	w.muxAgent.Lock()
 	defer w.muxAgent.Unlock()
 
@@ -247,11 +265,11 @@ func (w *WorkerICE) SessionID() ICESessionID {
 // will block until connection succeeded
 // but it won't release if ICE Agent went into Disconnected or Failed state,
 // so we have to cancel it with the provided context once agent detected a broken connection
-func (w *WorkerICE) connect(ctx context.Context, agent *icemaker.ThreadSafeAgent, remoteOfferAnswer *OfferAnswer) {
+func (w *ICE) connect(ctx context.Context, dialerCancel context.CancelFunc, agent *icemaker.ThreadSafeAgent, remoteOfferAnswer *signaling.OfferAnswer) {
 	w.log.Debugf("gather candidates")
 	if err := agent.GatherCandidates(); err != nil {
 		w.log.Warnf("failed to gather candidates: %s", err)
-		w.closeAgent(agent, w.agentDialerCancel)
+		w.closeAgent(agent, dialerCancel)
 		return
 	}
 
@@ -259,19 +277,19 @@ func (w *WorkerICE) connect(ctx context.Context, agent *icemaker.ThreadSafeAgent
 	remoteConn, err := w.turnAgentDial(ctx, agent, remoteOfferAnswer)
 	if err != nil {
 		w.log.Debugf("failed to dial the remote peer: %s", err)
-		w.closeAgent(agent, w.agentDialerCancel)
+		w.closeAgent(agent, dialerCancel)
 		return
 	}
 	w.log.Debugf("agent dial succeeded")
 
 	pair, err := agent.GetSelectedCandidatePair()
 	if err != nil {
-		w.closeAgent(agent, w.agentDialerCancel)
+		w.closeAgent(agent, dialerCancel)
 		return
 	}
 	if pair == nil {
 		w.log.Warnf("selected candidate pair is nil, cannot proceed")
-		w.closeAgent(agent, w.agentDialerCancel)
+		w.closeAgent(agent, dialerCancel)
 		return
 	}
 
@@ -299,17 +317,22 @@ func (w *WorkerICE) connect(ctx context.Context, agent *icemaker.ThreadSafeAgent
 	}
 	w.log.Debugf("on ICE conn is ready to use")
 
-	w.log.Infof("connection succeeded with offer session: %s", remoteOfferAnswer.SessionIDString())
 	w.muxAgent.Lock()
+	if w.agent != agent {
+		w.muxAgent.Unlock()
+		w.log.Debugf("agent has been replaced during connect, dropping obsolete connection")
+		return
+	}
 	w.agentConnecting = false
 	w.lastSuccess = time.Now()
+	w.connectedAgent = agent
 	w.muxAgent.Unlock()
 
-	// todo: the potential problem is a race between the onConnectionStateChange
-	w.conn.onICEConnectionIsReady(selectedPriority(pair), ci)
+	w.log.Infof("connection succeeded with offer session: %s", remoteOfferAnswer.SessionIDString())
+	w.onConnReady(selectedPriority(pair), ci)
 }
 
-func (w *WorkerICE) closeAgent(agent *icemaker.ThreadSafeAgent, cancel context.CancelFunc) bool {
+func (w *ICE) closeAgent(agent *icemaker.ThreadSafeAgent, cancel context.CancelFunc) bool {
 	cancel()
 	if err := agent.Close(); err != nil {
 		w.log.Warnf("failed to close ICE agent: %s", err)
@@ -323,7 +346,7 @@ func (w *WorkerICE) closeAgent(agent *icemaker.ThreadSafeAgent, cancel context.C
 
 	if w.agent == agent {
 		// consider to remove from here and move to the OnNewOffer
-		sessionID, err := NewICESessionID()
+		sessionID, err := icemaker.NewSessionID()
 		if err != nil {
 			w.log.Errorf("failed to create new session ID: %s", err)
 		}
@@ -335,7 +358,7 @@ func (w *WorkerICE) closeAgent(agent *icemaker.ThreadSafeAgent, cancel context.C
 	return sessionChanged
 }
 
-func (w *WorkerICE) punchRemoteWGPort(pair *ice.CandidatePair, remoteWgPort int) {
+func (w *ICE) punchRemoteWGPort(pair *ice.CandidatePair, remoteWgPort int) {
 	// wait local endpoint configuration
 	time.Sleep(time.Second)
 	addr, err := net.ResolveUDPAddr("udp", net.JoinHostPort(pair.Remote.Address(), strconv.Itoa(remoteWgPort)))
@@ -344,7 +367,7 @@ func (w *WorkerICE) punchRemoteWGPort(pair *ice.CandidatePair, remoteWgPort int)
 		return
 	}
 
-	mux, ok := w.config.ICEConfig.UDPMuxSrflx.(*udpmux.UniversalUDPMuxDefault)
+	mux, ok := w.iceConfig.UDPMuxSrflx.(*udpmux.UniversalUDPMuxDefault)
 	if !ok {
 		w.log.Warn("invalid udp mux conversion")
 		return
@@ -357,7 +380,7 @@ func (w *WorkerICE) punchRemoteWGPort(pair *ice.CandidatePair, remoteWgPort int)
 
 // onICECandidate is a callback attached to an ICE Agent to receive new local connection candidates
 // and then signals them to the remote peer
-func (w *WorkerICE) onICECandidate(candidate ice.Candidate) {
+func (w *ICE) onICECandidate(candidate ice.Candidate) {
 	// nil means candidate gathering has been ended
 	if candidate == nil {
 		return
@@ -366,9 +389,9 @@ func (w *WorkerICE) onICECandidate(candidate ice.Candidate) {
 	// TODO: reported port is incorrect for CandidateTypeHost, makes understanding ICE use via logs confusing as port is ignored
 	w.log.Debugf("discovered local candidate %s", candidate.String())
 	go func() {
-		err := w.signaler.SignalICECandidate(candidate, w.config.Key)
+		err := w.signaler.SignalICECandidate(candidate, w.key)
 		if err != nil {
-			w.log.Errorf("failed signaling candidate to the remote peer %s %s", w.config.Key, err)
+			w.log.Errorf("failed signaling candidate to the remote peer %s %s", w.key, err)
 		}
 	}()
 
@@ -378,8 +401,8 @@ func (w *WorkerICE) onICECandidate(candidate ice.Candidate) {
 }
 
 // injectPortForwardedCandidate signals an additional candidate using the pre-created port mapping.
-func (w *WorkerICE) injectPortForwardedCandidate(srflxCandidate ice.Candidate) {
-	pfManager := w.conn.portForwardManager
+func (w *ICE) injectPortForwardedCandidate(srflxCandidate ice.Candidate) {
+	pfManager := w.portForwardManager
 	if pfManager == nil {
 		return
 	}
@@ -407,7 +430,7 @@ func (w *WorkerICE) injectPortForwardedCandidate(srflxCandidate ice.Candidate) {
 		forwardedCandidate.String(), mapping.InternalPort, mapping.ExternalPort, mapping.NATType, forwardedCandidate.Priority())
 
 	go func() {
-		if err := w.signaler.SignalICECandidate(forwardedCandidate, w.config.Key); err != nil {
+		if err := w.signaler.SignalICECandidate(forwardedCandidate, w.key); err != nil {
 			w.log.Errorf("signal port-forwarded candidate: %v", err)
 		}
 	}()
@@ -415,7 +438,7 @@ func (w *WorkerICE) injectPortForwardedCandidate(srflxCandidate ice.Candidate) {
 
 // createForwardedCandidate creates a new server reflexive candidate with the forwarded port.
 // It uses the NAT gateway's external IP with the forwarded port.
-func (w *WorkerICE) createForwardedCandidate(srflxCandidate ice.Candidate, mapping *portforward.Mapping) (ice.Candidate, error) {
+func (w *ICE) createForwardedCandidate(srflxCandidate ice.Candidate, mapping *portforward.Mapping) (ice.Candidate, error) {
 	var externalIP string
 	if mapping.ExternalIP != nil && !mapping.ExternalIP.IsUnspecified() {
 		externalIP = mapping.ExternalIP.String()
@@ -460,9 +483,9 @@ func (w *WorkerICE) createForwardedCandidate(srflxCandidate ice.Candidate, mappi
 	return candidate, nil
 }
 
-func (w *WorkerICE) onICESelectedCandidatePair(agent *icemaker.ThreadSafeAgent, c1, c2 ice.Candidate) {
+func (w *ICE) onICESelectedCandidatePair(agent *icemaker.ThreadSafeAgent, c1, c2 ice.Candidate) {
 	w.log.Debugf("selected candidate pair [local <-> remote] -> [%s <-> %s], peer %s", c1.String(), c2.String(),
-		w.config.Key)
+		w.key)
 
 	pairStat, ok := agent.GetSelectedCandidatePairStats()
 	if !ok {
@@ -471,14 +494,14 @@ func (w *WorkerICE) onICESelectedCandidatePair(agent *icemaker.ThreadSafeAgent,
 	}
 
 	duration := time.Duration(pairStat.CurrentRoundTripTime * float64(time.Second))
-	if err := w.statusRecorder.UpdateLatency(w.config.Key, duration); err != nil {
+	if err := w.statusRecorder.UpdateLatency(w.key, duration); err != nil {
 		w.log.Debugf("failed to update latency for peer: %s", err)
 		return
 	}
 }
 
-func (w *WorkerICE) logSuccessfulPaths(agent *icemaker.ThreadSafeAgent) {
-	sessionID := w.SessionID()
+func (w *ICE) logSuccessfulPaths(agent *icemaker.ThreadSafeAgent) {
+	sessionID := w.getSessionID()
 	stats := agent.GetCandidatePairsStats()
 	localCandidates, _ := agent.GetLocalCandidates()
 	remoteCandidates, _ := agent.GetRemoteCandidates()
@@ -508,32 +531,44 @@ func (w *WorkerICE) logSuccessfulPaths(agent *icemaker.ThreadSafeAgent) {
 	}
 }
 
-func (w *WorkerICE) onConnectionStateChange(agent *icemaker.ThreadSafeAgent, dialerCancel context.CancelFunc) func(ice.ConnectionState) {
+func (w *ICE) onConnectionStateChange(agent *icemaker.ThreadSafeAgent, dialerCancel context.CancelFunc) func(ice.ConnectionState) {
+	// per-agent state; pion delivers callbacks of one agent sequentially
+	var connected bool
 	return func(state ice.ConnectionState) {
 		w.log.Debugf("ICE ConnectionState has changed to %s", state.String())
 		switch state {
 		case ice.ConnectionStateConnected:
-			w.lastKnownState = ice.ConnectionStateConnected
+			connected = true
 			w.logSuccessfulPaths(agent)
-			return
 		case ice.ConnectionStateFailed, ice.ConnectionStateDisconnected, ice.ConnectionStateClosed:
 			// ice.ConnectionStateClosed happens when we recreate the agent. For the P2P to TURN switch important to
 			// notify the conn.onICEStateDisconnected changes to update the current used priority
 
 			sessionChanged := w.closeAgent(agent, dialerCancel)
 
-			if w.lastKnownState == ice.ConnectionStateConnected {
-				w.lastKnownState = ice.ConnectionStateDisconnected
-				w.conn.onICEStateDisconnected(sessionChanged)
+			if !connected {
+				return
 			}
-		default:
-			return
+			connected = false
+
+			w.muxAgent.Lock()
+			stale := w.connectedAgent != agent
+			if !stale {
+				w.connectedAgent = nil
+			}
+			w.muxAgent.Unlock()
+
+			if stale {
+				w.log.Debugf("suppress disconnected event of replaced ICE agent")
+				return
+			}
+			w.onStatusDisconnect(sessionChanged)
 		}
 	}
 }
 
-func (w *WorkerICE) turnAgentDial(ctx context.Context, agent *icemaker.ThreadSafeAgent, remoteOfferAnswer *OfferAnswer) (*ice.Conn, error) {
-	if isController(w.config) {
+func (w *ICE) turnAgentDial(ctx context.Context, agent *icemaker.ThreadSafeAgent, remoteOfferAnswer *signaling.OfferAnswer) (*ice.Conn, error) {
+	if w.isController {
 		return agent.Dial(ctx, remoteOfferAnswer.IceCredentials.UFrag, remoteOfferAnswer.IceCredentials.Pwd)
 	} else {
 		return agent.Accept(ctx, remoteOfferAnswer.IceCredentials.UFrag, remoteOfferAnswer.IceCredentials.Pwd)
@@ -595,10 +630,10 @@ func isRelayed(pair *ice.CandidatePair) bool {
 	return false
 }
 
-func selectedPriority(pair *ice.CandidatePair) conntype.ConnPriority {
+func selectedPriority(pair *ice.CandidatePair) ConnPriority {
 	if isRelayed(pair) {
-		return conntype.ICETurn
+		return ICETurn
 	} else {
-		return conntype.ICEP2P
+		return ICEP2P
 	}
 }

client/internal/peer/worker/worker_relay.go

@@ -1,4 +1,4 @@
-package peer
+package worker
 
 import (
 	"context"
@@ -10,22 +10,23 @@ import (
 
 	log "github.com/sirupsen/logrus"
 
+	"github.com/netbirdio/netbird/client/internal/peer/signaling"
 	relayClient "github.com/netbirdio/netbird/shared/relay/client"
 )
 
 type RelayConnInfo struct {
-	relayedConn     net.Conn
-	rosenpassPubKey []byte
-	rosenpassAddr   string
+	RelayedConn     net.Conn
+	RosenpassPubKey []byte
+	RosenpassAddr   string
 }
 
 type WorkerRelay struct {
-	peerCtx      context.Context
-	log          *log.Entry
-	isController bool
-	config       ConnConfig
-	conn         *Conn
-	relayManager *relayClient.Manager
+	log            *log.Entry
+	key            string
+	isController   bool
+	onConnReady    func(RelayConnInfo)
+	onDisconnected func()
+	relayManager   *relayClient.Manager
 
 	relayedConn net.Conn
 	relayLock   sync.Mutex
@@ -33,19 +34,19 @@ type WorkerRelay struct {
 	relaySupportedOnRemotePeer atomic.Bool
 }
 
-func NewWorkerRelay(ctx context.Context, log *log.Entry, ctrl bool, config ConnConfig, conn *Conn, relayManager *relayClient.Manager) *WorkerRelay {
+func NewWorkerRelay(log *log.Entry, key string, isController bool, onConnReady func(RelayConnInfo), onDisconnected func(), relayManager *relayClient.Manager) *WorkerRelay {
 	r := &WorkerRelay{
-		peerCtx:      ctx,
-		log:          log,
-		isController: ctrl,
-		config:       config,
-		conn:         conn,
-		relayManager: relayManager,
+		log:            log,
+		key:            key,
+		isController:   isController,
+		onConnReady:    onConnReady,
+		onDisconnected: onDisconnected,
+		relayManager:   relayManager,
 	}
 	return r
 }
 
-func (w *WorkerRelay) OnNewOffer(remoteOfferAnswer *OfferAnswer) {
+func (w *WorkerRelay) OnNewOffer(ctx context.Context, remoteOfferAnswer *signaling.OfferAnswer) {
 	if !w.isRelaySupported(remoteOfferAnswer) {
 		w.log.Infof("Relay is not supported by remote peer")
 		w.relaySupportedOnRemotePeer.Store(false)
@@ -66,7 +67,7 @@ func (w *WorkerRelay) OnNewOffer(remoteOfferAnswer *OfferAnswer) {
 		serverIP = remoteOfferAnswer.RelaySrvIP
 	}
 
-	relayedConn, err := w.relayManager.OpenConn(w.peerCtx, srv, w.config.Key, serverIP)
+	relayedConn, err := w.relayManager.OpenConn(ctx, srv, w.key, serverIP)
 	if err != nil {
 		if errors.Is(err, relayClient.ErrConnAlreadyExists) {
 			w.log.Debugf("handled offer by reusing existing relay connection")
@@ -88,10 +89,10 @@ func (w *WorkerRelay) OnNewOffer(remoteOfferAnswer *OfferAnswer) {
 	}
 
 	w.log.Debugf("peer conn opened via Relay: %s", srv)
-	go w.conn.onRelayConnectionIsReady(RelayConnInfo{
-		relayedConn:     relayedConn,
-		rosenpassPubKey: remoteOfferAnswer.RosenpassPubKey,
-		rosenpassAddr:   remoteOfferAnswer.RosenpassAddr,
+	w.onConnReady(RelayConnInfo{
+		RelayedConn:     relayedConn,
+		RosenpassPubKey: remoteOfferAnswer.RosenpassPubKey,
+		RosenpassAddr:   remoteOfferAnswer.RosenpassAddr,
 	})
 }
 
@@ -119,7 +120,7 @@ func (w *WorkerRelay) CloseConn() {
 	}
 }
 
-func (w *WorkerRelay) isRelaySupported(answer *OfferAnswer) bool {
+func (w *WorkerRelay) isRelaySupported(answer *signaling.OfferAnswer) bool {
 	if !w.relayManager.HasRelayAddress() {
 		return false
 	}
@@ -134,5 +135,5 @@ func (w *WorkerRelay) preferredRelayServer(myRelayAddress, remoteRelayAddress st
 }
 
 func (w *WorkerRelay) onRelayClientDisconnected() {
-	go w.conn.onRelayDisconnected()
+	w.onDisconnected()
 }

client/internal/peer/worker_status.go

@@ -1,4 +1,4 @@
-package worker
+package peer
 
 import (
 	"sync/atomic"
@@ -7,17 +7,17 @@ import (
 )
 
 const (
-	StatusDisconnected Status = iota
-	StatusConnected
+	WorkerStatusDisconnected WorkerStatus = iota
+	WorkerStatusConnected
 )
 
-type Status int32
+type WorkerStatus int32
 
-func (s Status) String() string {
+func (s WorkerStatus) String() string {
 	switch s {
-	case StatusDisconnected:
+	case WorkerStatusDisconnected:
 		return "Disconnected"
-	case StatusConnected:
+	case WorkerStatusConnected:
 		return "Connected"
 	default:
 		log.Errorf("unknown status: %d", s)
@@ -37,16 +37,16 @@ func NewAtomicStatus() *AtomicWorkerStatus {
 }
 
 // Get returns the current connection status
-func (acs *AtomicWorkerStatus) Get() Status {
-	return Status(acs.status.Load())
+func (acs *AtomicWorkerStatus) Get() WorkerStatus {
+	return WorkerStatus(acs.status.Load())
 }
 
 func (acs *AtomicWorkerStatus) SetConnected() {
-	acs.status.Store(int32(StatusConnected))
+	acs.status.Store(int32(WorkerStatusConnected))
 }
 
 func (acs *AtomicWorkerStatus) SetDisconnected() {
-	acs.status.Store(int32(StatusDisconnected))
+	acs.status.Store(int32(WorkerStatusDisconnected))
 }
 
 // String returns the string representation of the current status

client/internal/profilemanager/config.go

@@ -101,8 +101,6 @@ type ConfigInput struct {
 
 	DNSLabels domain.List
 
-	LazyConnectionEnabled *bool
-
 	MTU *uint16
 }
 
@@ -180,7 +178,9 @@ type Config struct {
 
 	ClientCertKeyPair *tls.Certificate `json:"-"`
 
-	LazyConnectionEnabled bool
+	// LazyConnection is the MDM-managed lazy-connection override ("on"/"off"/"").
+	// Runtime-only: re-derived from MDM policy on each load, never persisted.
+	LazyConnection string `json:"-"`
 
 	MTU uint16
 
@@ -632,12 +632,6 @@ func (config *Config) apply(input ConfigInput) (updated bool, err error) {
 		updated = true
 	}
 
-	if input.LazyConnectionEnabled != nil && *input.LazyConnectionEnabled != config.LazyConnectionEnabled {
-		log.Infof("switching lazy connection to %t", *input.LazyConnectionEnabled)
-		config.LazyConnectionEnabled = *input.LazyConnectionEnabled
-		updated = true
-	}
-
 	if input.MTU != nil && *input.MTU != config.MTU {
 		log.Infof("updating MTU to %d (old value %d)", *input.MTU, config.MTU)
 		config.MTU = *input.MTU
@@ -728,6 +722,15 @@ func (config *Config) applyMDMPolicy(policy *mdm.Policy) {
 			log.Warnf("MDM wireguard port %d out of range [1,65535]; keeping previous value", v)
 		}
 	}
+
+	if v, ok := policy.GetBool(mdm.KeyLazyConnection); ok {
+		state := "off"
+		if v {
+			state = "on"
+		}
+		config.LazyConnection = state
+		logApplied(mdm.KeyLazyConnection, state)
+	}
 }
 
 // parseURL parses and validates the URL for the named service. The URL

client/internal/profilemanager/config_mdm_test.go

@@ -130,6 +130,37 @@ func TestApply_MDMBoolKeysOverrideOnDiskValue(t *testing.T) {
 	assert.True(t, cfg.Policy().HasKey(mdm.KeyRosenpassEnabled))
 }
 
+func TestApply_MDMLazyConnection(t *testing.T) {
+	cases := []struct {
+		name string
+		raw  any
+		want string
+	}{
+		{"native true", true, "on"},
+		{"native false", false, "off"},
+		{"string on", "on", "on"},
+		{"string off", "off", "off"},
+		{"string yes", "yes", "on"},
+		{"string no", "no", "off"},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			withMDMPolicy(t, mdm.NewPolicy(map[string]any{
+				mdm.KeyLazyConnection: c.raw,
+			}))
+
+			cfg, err := UpdateOrCreateConfig(ConfigInput{
+				ConfigPath: filepath.Join(t.TempDir(), "config.json"),
+			})
+			require.NoError(t, err)
+			require.NotNil(t, cfg)
+
+			assert.Equal(t, c.want, cfg.LazyConnection)
+			assert.True(t, cfg.Policy().HasKey(mdm.KeyLazyConnection))
+		})
+	}
+}
+
 func TestApply_MDMPreSharedKeyRedactionSentinelRejected(t *testing.T) {
 	const maskSentinel = "**********"
 

client/internal/profilemanager/service.go

@@ -11,6 +11,7 @@ import (
 	"runtime"
 	"sort"
 	"strings"
+	"syscall"
 
 	log "github.com/sirupsen/logrus"
 
@@ -439,7 +440,11 @@ func (s *ServiceManager) GetStatePath() string {
 
 	activeProf, err := s.GetActiveProfileState()
 	if err != nil {
-		log.Warnf("failed to get active profile state: %v", err)
+		if errors.Is(err, syscall.ENOSYS) {
+			log.Debugf("active profile state unavailable on this platform: %v", err)
+		} else {
+			log.Warnf("failed to get active profile state: %v", err)
+		}
 		return defaultStatePath
 	}
 

client/internal/rosenpass/manager.go

@@ -39,6 +39,7 @@ type rpServer interface {
 
 type Manager struct {
 	ifaceName    string
+	localWgKey   wgtypes.Key
 	spk          []byte
 	ssk          []byte
 	rpKeyHash    string
@@ -51,8 +52,9 @@ type Manager struct {
 	wgIface      PresharedKeySetter
 }
 
-// NewManager creates a new Rosenpass manager
-func NewManager(preSharedKey *wgtypes.Key, wgIfaceName string) (*Manager, error) {
+// NewManager creates a new Rosenpass manager. localWgKey is the local
+// WireGuard public key, used to derive the per-peer rendezvous key.
+func NewManager(preSharedKey *wgtypes.Key, wgIfaceName string, localWgKey wgtypes.Key) (*Manager, error) {
 	public, secret, err := rp.GenerateKeyPair()
 	if err != nil {
 		return nil, err
@@ -62,6 +64,7 @@ func NewManager(preSharedKey *wgtypes.Key, wgIfaceName string) (*Manager, error)
 	log.Tracef("generated new rosenpass key pair with public key %s", rpKeyHash)
 	return &Manager{
 		ifaceName:    wgIfaceName,
+		localWgKey:   localWgKey,
 		rpKeyHash:    rpKeyHash,
 		spk:          public,
 		ssk:          secret,
@@ -73,7 +76,7 @@ func NewManager(preSharedKey *wgtypes.Key, wgIfaceName string) (*Manager, error)
 		// nil receiver in addPeer -> m.rpWgHandler.AddPeer. generateConfig will
 		// replace it with a fresh handler on each Run() to clear stale peer
 		// state from previous engine sessions.
-		rpWgHandler: NewNetbirdHandler(),
+		rpWgHandler: NewNetbirdHandler((*[32]byte)(preSharedKey), localWgKey),
 		lock:        sync.Mutex{},
 	}, nil
 }
@@ -161,7 +164,7 @@ func (m *Manager) generateConfig() (rp.Config, error) {
 	cfg.Peers = []rp.PeerConfig{}
 
 	m.lock.Lock()
-	m.rpWgHandler = NewNetbirdHandler()
+	m.rpWgHandler = NewNetbirdHandler(m.preSharedKey, m.localWgKey)
 	if m.wgIface != nil {
 		m.rpWgHandler.SetInterface(m.wgIface)
 	}

client/internal/rosenpass/manager_test.go

@@ -85,7 +85,7 @@ func newTestManager(spkFirstByte byte, mock *mockServer) *Manager {
 		ssk:         make([]byte, 32),
 		rpKeyHash:   "test-hash",
 		rpPeerIDs:   make(map[string]*rp.PeerID),
-		rpWgHandler: NewNetbirdHandler(),
+		rpWgHandler: NewNetbirdHandler(nil, wgtypes.Key{0x01}),
 		server:      mock,
 	}
 }
@@ -255,7 +255,7 @@ func TestAddPeer_NilServer_ReturnsErrorNoCrash(t *testing.T) {
 // issue #4341 cannot occur in the window between NewManager and Run().
 func TestNewManager_PreInitializesHandler(t *testing.T) {
 	psk := wgtypes.Key{}
-	m, err := NewManager(&psk, "wt0")
+	m, err := NewManager(&psk, "wt0", wgtypes.Key{0x01})
 	require.NoError(t, err)
 	require.NotNil(t, m.rpWgHandler, "rpWgHandler must be initialized in NewManager")
 }
@@ -329,10 +329,10 @@ func TestIsPresharedKeyInitialized_AddedButNotHandshaken_ReturnsFalse(t *testing
 	require.False(t, m.IsPresharedKeyInitialized(wgKey))
 }
 
-// --- NetbirdHandler.outputKey ----------------------------------------------
+// --- NetbirdHandler.applyKey ----------------------------------------------
 
-func TestHandler_OutputKey_FirstCallUsesUpdateOnlyFalse(t *testing.T) {
-	h := NewNetbirdHandler()
+func TestHandler_ApplyKey_FirstCallUsesUpdateOnlyFalse(t *testing.T) {
+	h := NewNetbirdHandler(nil, wgtypes.Key{0x01})
 	iface := &mockIface{}
 	h.SetInterface(iface)
 
@@ -348,8 +348,8 @@ func TestHandler_OutputKey_FirstCallUsesUpdateOnlyFalse(t *testing.T) {
 	require.Equal(t, wgKey.String(), iface.calls[0].peerKey)
 }
 
-func TestHandler_OutputKey_SubsequentCallsUseUpdateOnlyTrue(t *testing.T) {
-	h := NewNetbirdHandler()
+func TestHandler_ApplyKey_SubsequentCallsUseUpdateOnlyTrue(t *testing.T) {
+	h := NewNetbirdHandler(nil, wgtypes.Key{0x01})
 	iface := &mockIface{}
 	h.SetInterface(iface)
 
@@ -364,8 +364,8 @@ func TestHandler_OutputKey_SubsequentCallsUseUpdateOnlyTrue(t *testing.T) {
 	require.True(t, iface.calls[1].updateOnly, "subsequent rotations must use updateOnly=true")
 }
 
-func TestHandler_OutputKey_NilInterface_NoCrashNoCall(t *testing.T) {
-	h := NewNetbirdHandler()
+func TestHandler_ApplyKey_NilInterface_NoCrashNoCall(t *testing.T) {
+	h := NewNetbirdHandler(nil, wgtypes.Key{0x01})
 	// no SetInterface — iface remains nil
 	pid := rp.PeerID{0x03}
 	h.AddPeer(pid, "wt0", rp.Key(wgtypes.Key{}))
@@ -374,8 +374,8 @@ func TestHandler_OutputKey_NilInterface_NoCrashNoCall(t *testing.T) {
 	h.HandshakeCompleted(pid, rp.Key{})
 }
 
-func TestHandler_OutputKey_UnknownPeer_NoCall(t *testing.T) {
-	h := NewNetbirdHandler()
+func TestHandler_ApplyKey_UnknownPeer_NoCall(t *testing.T) {
+	h := NewNetbirdHandler(nil, wgtypes.Key{0x01})
 	iface := &mockIface{}
 	h.SetInterface(iface)
 
@@ -384,7 +384,7 @@ func TestHandler_OutputKey_UnknownPeer_NoCall(t *testing.T) {
 }
 
 func TestHandler_RemovePeer_ClearsInitializedState(t *testing.T) {
-	h := NewNetbirdHandler()
+	h := NewNetbirdHandler(nil, wgtypes.Key{0x01})
 	iface := &mockIface{}
 	h.SetInterface(iface)
 
@@ -398,7 +398,7 @@ func TestHandler_RemovePeer_ClearsInitializedState(t *testing.T) {
 }
 
 func TestHandler_SetInterfaceAfterAddPeer_StillReceivesKey(t *testing.T) {
-	h := NewNetbirdHandler()
+	h := NewNetbirdHandler(nil, wgtypes.Key{0x01})
 	pid := rp.PeerID{0x05}
 	wgKey := wgtypes.Key{0xEE}
 	h.AddPeer(pid, "wt0", rp.Key(wgKey))

client/internal/rosenpass/netbird_handler.go

@@ -18,19 +18,34 @@ type PresharedKeySetter interface {
 type wireGuardPeer struct {
 	Interface string
 	PublicKey rp.Key
+	// initialized is true once a completed exchange has set a
+	// Rosenpass-managed PSK for this peer.
+	initialized bool
+	// chainKey is the key output by the last completed exchange, advanced by
+	// one ratchet step on expiry. Nil until the first exchange completes and
+	// after the peer has fallen back to the rendezvous key.
+	chainKey *wgtypes.Key
+	// expiries counts failed renewals since the last completed exchange.
+	expiries int
 }
 
 type NetbirdHandler struct {
-	mu               sync.Mutex
-	iface            PresharedKeySetter
-	peers            map[rp.PeerID]wireGuardPeer
-	initializedPeers map[rp.PeerID]bool
+	mu    sync.Mutex
+	iface PresharedKeySetter
+	// preSharedKey is the account-level preshared key, used as the rendezvous
+	// key when set. Nil means the deterministic seed key is used instead.
+	preSharedKey *[32]byte
+	// localWgKey is the local WireGuard public key, one of the two inputs to
+	// the deterministic seed key.
+	localWgKey wgtypes.Key
+	peers      map[rp.PeerID]*wireGuardPeer
 }
 
-func NewNetbirdHandler() *NetbirdHandler {
+func NewNetbirdHandler(preSharedKey *[32]byte, localWgKey wgtypes.Key) *NetbirdHandler {
 	return &NetbirdHandler{
-		peers:            map[rp.PeerID]wireGuardPeer{},
-		initializedPeers: map[rp.PeerID]bool{},
+		preSharedKey: preSharedKey,
+		localWgKey:   localWgKey,
+		peers:        map[rp.PeerID]*wireGuardPeer{},
 	}
 }
 
@@ -42,10 +57,16 @@ func (h *NetbirdHandler) SetInterface(iface PresharedKeySetter) {
 	h.iface = iface
 }
 
+// AddPeer registers a peer with the handler. Re-adding a known peer (every
+// reconnection does) keeps its key recovery state.
 func (h *NetbirdHandler) AddPeer(pid rp.PeerID, intf string, pk rp.Key) {
 	h.mu.Lock()
 	defer h.mu.Unlock()
-	h.peers[pid] = wireGuardPeer{
+	if existing, ok := h.peers[pid]; ok && existing.PublicKey == pk {
+		existing.Interface = intf
+		return
+	}
+	h.peers[pid] = &wireGuardPeer{
 		Interface: intf,
 		PublicKey: pk,
 	}
@@ -55,7 +76,6 @@ func (h *NetbirdHandler) RemovePeer(pid rp.PeerID) {
 	h.mu.Lock()
 	defer h.mu.Unlock()
 	delete(h.peers, pid)
-	delete(h.initializedPeers, pid)
 }
 
 // IsPeerInitialized returns true if Rosenpass has completed a handshake
@@ -63,50 +83,120 @@ func (h *NetbirdHandler) RemovePeer(pid rp.PeerID) {
 func (h *NetbirdHandler) IsPeerInitialized(pid rp.PeerID) bool {
 	h.mu.Lock()
 	defer h.mu.Unlock()
-	return h.initializedPeers[pid]
+	peer, ok := h.peers[pid]
+	return ok && peer.initialized
 }
 
+// HandshakeCompleted programs the freshly exchanged output key and resets the
+// peer's key recovery state.
 func (h *NetbirdHandler) HandshakeCompleted(pid rp.PeerID, key rp.Key) {
-	h.outputKey(rp.KeyOutputReasonStale, pid, key)
-}
+	psk := wgtypes.Key(key)
 
-func (h *NetbirdHandler) HandshakeExpired(pid rp.PeerID) {
-	key, _ := rp.GeneratePresharedKey()
-	h.outputKey(rp.KeyOutputReasonStale, pid, key)
-}
-
-func (h *NetbirdHandler) outputKey(_ rp.KeyOutputReason, pid rp.PeerID, psk rp.Key) {
 	h.mu.Lock()
-	iface := h.iface
-	wg, ok := h.peers[pid]
-	isInitialized := h.initializedPeers[pid]
-	h.mu.Unlock()
+	defer h.mu.Unlock()
 
-	if iface == nil {
-		log.Warn("rosenpass: interface not set, cannot update preshared key")
+	peer, ok := h.peers[pid]
+	if !ok {
 		return
 	}
+	if peer.expiries > 0 {
+		log.Infof("rosenpass exchange completed for peer %s after %d expired renewals", wgtypes.Key(peer.PublicKey), peer.expiries)
+	}
+	// chainKey tracks the shared exchange output regardless of the local write
+	// outcome, so both ends still converge on the next expiry.
+	peer.chainKey = &psk
+	peer.expiries = 0
+	if !h.applyKeyLocked(pid, psk, peer.initialized) {
+		return
+	}
+	peer.initialized = true
+}
 
+// HandshakeExpired replaces the expired key. The renewal exchange runs over
+// the tunnel keyed by the PSK itself, so the replacement must be derivable on
+// both ends without communication: the first expiry ratchets the last shared
+// key forward, repeated expiries (and expiries without a completed exchange)
+// fall back to the rendezvous key and drop the peer out of the initialized
+// state so connection reconfigurations reprogram the rendezvous key as well.
+func (h *NetbirdHandler) HandshakeExpired(pid rp.PeerID) {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+
+	peer, ok := h.peers[pid]
 	if !ok {
 		return
 	}
 
-	peerKey := wgtypes.Key(wg.PublicKey).String()
-	pskKey := wgtypes.Key(psk)
+	peer.expiries++
 
-	// Use updateOnly=true for later rotations (peer already has Rosenpass PSK)
-	// Use updateOnly=false for first rotation (peer has original/empty PSK)
-	if err := iface.SetPresharedKey(peerKey, pskKey, isInitialized); err != nil {
+	var psk wgtypes.Key
+	if peer.chainKey != nil && peer.expiries == 1 {
+		log.Infof("rosenpass key for peer %s expired without renewal, advancing to ratcheted key", wgtypes.Key(peer.PublicKey))
+		psk = RatchetKey(*peer.chainKey)
+		peer.chainKey = &psk
+	} else {
+		rendezvous, err := h.rendezvousKey(peer)
+		if err != nil {
+			// Fail closed: without a rendezvous key the expired key must
+			// still be rotated out, even if the replacement is unusable.
+			log.Errorf("failed to derive rendezvous key, replacing expired key with a random one: %v", err)
+			h.applyRandomKeyLocked(pid)
+			return
+		}
+		log.Warnf("rosenpass key for peer %s expired %d times without renewal, falling back to the rendezvous key", wgtypes.Key(peer.PublicKey), peer.expiries)
+		psk = rendezvous
+		peer.chainKey = nil
+		peer.initialized = false
+	}
+
+	h.applyKeyLocked(pid, psk, true)
+}
+
+// rendezvousKey returns the key both ends converge on without communication:
+// the account-level preshared key when configured, the deterministic seed key
+// otherwise. It mirrors the key that peer connections program when Rosenpass
+// does not manage the peer yet.
+func (h *NetbirdHandler) rendezvousKey(peer *wireGuardPeer) (wgtypes.Key, error) {
+	if h.preSharedKey != nil {
+		return *h.preSharedKey, nil
+	}
+
+	seed, err := DeterministicSeedKey(h.localWgKey.String(), wgtypes.Key(peer.PublicKey).String())
+	if err != nil {
+		return wgtypes.Key{}, err
+	}
+	return *seed, nil
+}
+
+// applyKeyLocked writes the preshared key for the peer to the WireGuard
+// interface and reports whether the write succeeded. Callers must hold h.mu
+// for the whole state-mutation-plus-write so that a concurrent completion and
+// expiry cannot reorder their writes relative to the in-memory chain key.
+func (h *NetbirdHandler) applyKeyLocked(pid rp.PeerID, psk wgtypes.Key, updateOnly bool) bool {
+	peer, ok := h.peers[pid]
+	if !ok {
+		return false
+	}
+
+	if h.iface == nil {
+		log.Warn("rosenpass: interface not set, cannot update preshared key")
+		return false
+	}
+
+	peerKey := wgtypes.Key(peer.PublicKey).String()
+	if err := h.iface.SetPresharedKey(peerKey, psk, updateOnly); err != nil {
 		log.Errorf("Failed to apply rosenpass key: %v", err)
+		return false
+	}
+
+	return true
+}
+
+func (h *NetbirdHandler) applyRandomKeyLocked(pid rp.PeerID) {
+	key, err := rp.GeneratePresharedKey()
+	if err != nil {
+		log.Errorf("failed to generate random preshared key: %v", err)
 		return
 	}
-
-	// Mark peer as isInitialized after the successful first rotation
-	if !isInitialized {
-		h.mu.Lock()
-		if _, exists := h.peers[pid]; exists {
-			h.initializedPeers[pid] = true
-		}
-		h.mu.Unlock()
-	}
+	h.applyKeyLocked(pid, wgtypes.Key(key), true)
 }

client/internal/rosenpass/netbird_handler_test.go

@@ -0,0 +1,250 @@
+package rosenpass
+
+import (
+	"testing"
+
+	rp "cunicu.li/go-rosenpass"
+	"github.com/stretchr/testify/require"
+	"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
+)
+
+// handlerTestLink wires two NetbirdHandlers as the two ends of a single
+// tunnel: handler A manages the rosenpass peer B and vice versa, the way two
+// NetBird clients see each other.
+type handlerTestLink struct {
+	handlerA, handlerB *NetbirdHandler
+	ifaceA, ifaceB     *mockIface
+	pidA, pidB         rp.PeerID
+	wgKeyA, wgKeyB     wgtypes.Key
+}
+
+func newHandlerTestLink(t *testing.T, preSharedKey *[32]byte) *handlerTestLink {
+	t.Helper()
+
+	link := &handlerTestLink{
+		ifaceA: &mockIface{},
+		ifaceB: &mockIface{},
+	}
+	link.pidA[0] = 0xaa
+	link.pidB[0] = 0xbb
+	link.wgKeyA[31] = 1
+	link.wgKeyB[31] = 2
+
+	link.handlerA = NewNetbirdHandler(preSharedKey, link.wgKeyA)
+	link.handlerB = NewNetbirdHandler(preSharedKey, link.wgKeyB)
+
+	link.handlerA.SetInterface(link.ifaceA)
+	link.handlerB.SetInterface(link.ifaceB)
+
+	link.handlerA.AddPeer(link.pidB, "wt0", rp.Key(link.wgKeyB))
+	link.handlerB.AddPeer(link.pidA, "wt0", rp.Key(link.wgKeyA))
+
+	return link
+}
+
+// complete simulates a completed rosenpass exchange: both ends derive the
+// same output key.
+func (l *handlerTestLink) complete(osk rp.Key) {
+	l.handlerA.HandshakeCompleted(l.pidB, osk)
+	l.handlerB.HandshakeCompleted(l.pidA, osk)
+}
+
+// expire simulates a failed key renewal on both ends.
+func (l *handlerTestLink) expire() {
+	l.handlerA.HandshakeExpired(l.pidB)
+	l.handlerB.HandshakeExpired(l.pidA)
+}
+
+func lastPSK(t *testing.T, m *mockIface) wgtypes.Key {
+	t.Helper()
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	require.NotEmpty(t, m.calls, "expected at least one SetPresharedKey call")
+	return m.calls[len(m.calls)-1].psk
+}
+
+func TestHandshakeCompleted_SetsKeyAndInitializes(t *testing.T) {
+	link := newHandlerTestLink(t, nil)
+
+	var osk rp.Key
+	osk[0] = 0x42
+	link.complete(osk)
+
+	require.Equal(t, wgtypes.Key(osk), lastPSK(t, link.ifaceA), "completed exchange must program the osk")
+	require.False(t, link.ifaceA.calls[0].updateOnly, "first rotation must not be update-only")
+	require.True(t, link.handlerA.IsPeerInitialized(link.pidB), "peer must be initialized after first completed exchange")
+
+	link.complete(osk)
+	require.True(t, link.ifaceA.calls[1].updateOnly, "later rotations must be update-only")
+}
+
+// TestHandshakeExpired_BothSidesConverge encodes the core recovery invariant:
+// rosenpass renewals run over the tunnel that the PSK itself keys, so when a
+// renewal fails on both ends, both ends must fall back to the same key or the
+// tunnel can never handshake again.
+func TestHandshakeExpired_BothSidesConverge(t *testing.T) {
+	link := newHandlerTestLink(t, nil)
+
+	var osk rp.Key
+	osk[0] = 0x42
+	link.complete(osk)
+
+	link.expire()
+	keyA := lastPSK(t, link.ifaceA)
+	keyB := lastPSK(t, link.ifaceB)
+	require.NotEqual(t, wgtypes.Key(osk), keyA, "expired key must be rotated out")
+	require.Equal(t, keyA, keyB, "both ends must converge on the same key after expiry")
+
+	link.expire()
+	require.Equal(t, lastPSK(t, link.ifaceA), lastPSK(t, link.ifaceB),
+		"both ends must still converge after repeated expiries")
+}
+
+// TestHandshakeExpired_ExpiryWithoutCompletionConverges covers the bootstrap
+// case: the initial exchange never completed (the tunnel ran on the rendezvous
+// key), so an expiry must not replace the working key with an unrecoverable
+// one on either end.
+func TestHandshakeExpired_ExpiryWithoutCompletionConverges(t *testing.T) {
+	link := newHandlerTestLink(t, nil)
+
+	link.expire()
+	require.Equal(t, lastPSK(t, link.ifaceA), lastPSK(t, link.ifaceB),
+		"both ends must converge when the exchange never completed")
+}
+
+// TestHandshakeExpired_RepeatedExpiryClearsInitialized: once renewals keep
+// failing, the peer must drop out of the initialized state so the next
+// connection reconfiguration reprograms the rendezvous key instead of
+// preserving a poisoned rosenpass-managed key.
+func TestHandshakeExpired_RepeatedExpiryClearsInitialized(t *testing.T) {
+	link := newHandlerTestLink(t, nil)
+
+	var osk rp.Key
+	osk[0] = 0x42
+	link.complete(osk)
+
+	link.expire()
+	link.expire()
+
+	require.False(t, link.handlerA.IsPeerInitialized(link.pidB),
+		"repeated expiries must clear the initialized state")
+	require.False(t, link.handlerB.IsPeerInitialized(link.pidA),
+		"repeated expiries must clear the initialized state")
+}
+
+// TestHandshakeCompleted_AfterExpiryRecovers: a completed exchange after a
+// desync must fully reset the recovery state.
+func TestHandshakeCompleted_AfterExpiryRecovers(t *testing.T) {
+	link := newHandlerTestLink(t, nil)
+
+	var osk1, osk2 rp.Key
+	osk1[0] = 1
+	osk2[0] = 2
+
+	link.complete(osk1)
+	link.expire()
+	link.expire()
+
+	link.complete(osk2)
+	require.Equal(t, wgtypes.Key(osk2), lastPSK(t, link.ifaceA), "new exchange must program the fresh osk")
+	require.True(t, link.handlerA.IsPeerInitialized(link.pidB), "peer must be initialized again after recovery")
+
+	link.expire()
+	require.Equal(t, lastPSK(t, link.ifaceA), lastPSK(t, link.ifaceB),
+		"recovered link must converge again on the next expiry")
+	require.NotEqual(t, wgtypes.Key(osk2), lastPSK(t, link.ifaceA), "expired key must be rotated out")
+}
+
+// TestHandshakeExpired_FirstExpiryRatchetsLastKey: the first expiry must
+// derive the replacement from the last shared key, so an attacker who only
+// blocks the renewal exchange gains nothing over the previous key.
+func TestHandshakeExpired_FirstExpiryRatchetsLastKey(t *testing.T) {
+	link := newHandlerTestLink(t, nil)
+
+	var osk rp.Key
+	osk[0] = 0x42
+	link.complete(osk)
+
+	link.expire()
+	require.Equal(t, RatchetKey(wgtypes.Key(osk)), lastPSK(t, link.ifaceA),
+		"first expiry must program the ratcheted key")
+	require.True(t, link.handlerA.IsPeerInitialized(link.pidB),
+		"ratchet step must keep the peer initialized so reconfigurations preserve the key")
+}
+
+// TestHandshakeExpired_RepeatedExpiryFallsBackToSeed: once the ratchet key
+// also fails, both ends must land on the same key that peer connections
+// program for uninitialized peers, so a reconnect completes the recovery.
+func TestHandshakeExpired_RepeatedExpiryFallsBackToSeed(t *testing.T) {
+	link := newHandlerTestLink(t, nil)
+
+	var osk rp.Key
+	osk[0] = 0x42
+	link.complete(osk)
+
+	link.expire()
+	link.expire()
+
+	seed, err := DeterministicSeedKey(link.wgKeyA.String(), link.wgKeyB.String())
+	require.NoError(t, err)
+	require.Equal(t, *seed, lastPSK(t, link.ifaceA), "repeated expiry must fall back to the seed key")
+	require.Equal(t, *seed, lastPSK(t, link.ifaceB), "repeated expiry must fall back to the seed key")
+}
+
+// TestHandshakeExpired_ConfiguredPSKUsedAsRendezvous: with an account-level
+// preshared key configured, the fallback must be that key, matching what peer
+// connections program for uninitialized peers.
+func TestHandshakeExpired_ConfiguredPSKUsedAsRendezvous(t *testing.T) {
+	psk := &[32]byte{0x77}
+	link := newHandlerTestLink(t, psk)
+
+	var osk rp.Key
+	osk[0] = 0x42
+	link.complete(osk)
+
+	link.expire()
+	link.expire()
+
+	require.Equal(t, wgtypes.Key(*psk), lastPSK(t, link.ifaceA),
+		"fallback must be the configured preshared key")
+	require.Equal(t, wgtypes.Key(*psk), lastPSK(t, link.ifaceB),
+		"fallback must be the configured preshared key on both ends")
+}
+
+// TestHandshakeExpired_ExpiryWritesAreUpdateOnly: expiry replacements must
+// never create a WireGuard peer that connection management has removed.
+func TestHandshakeExpired_ExpiryWritesAreUpdateOnly(t *testing.T) {
+	link := newHandlerTestLink(t, nil)
+
+	var osk rp.Key
+	osk[0] = 0x42
+	link.complete(osk)
+
+	link.expire()
+	link.expire()
+
+	for _, call := range link.ifaceA.calls[1:] {
+		require.True(t, call.updateOnly, "expiry writes must be update-only")
+	}
+}
+
+// TestAddPeer_ReAddKeepsRecoveryState: reconnections re-add the peer on every
+// OnConnected; that must not reset the expiry chain state.
+func TestAddPeer_ReAddKeepsRecoveryState(t *testing.T) {
+	link := newHandlerTestLink(t, nil)
+
+	var osk rp.Key
+	osk[0] = 0x42
+	link.complete(osk)
+	link.expire()
+
+	link.handlerA.AddPeer(link.pidB, "wt0", rp.Key(link.wgKeyB))
+	require.True(t, link.handlerA.IsPeerInitialized(link.pidB),
+		"re-adding a known peer must keep its state")
+
+	link.expire()
+	seed, err := DeterministicSeedKey(link.wgKeyA.String(), link.wgKeyB.String())
+	require.NoError(t, err)
+	require.Equal(t, *seed, lastPSK(t, link.ifaceA),
+		"second expiry after re-add must continue to the seed fallback")
+}

client/internal/rosenpass/seed.go

@@ -1,11 +1,28 @@
 package rosenpass
 
 import (
+	"crypto/sha256"
 	"fmt"
 
 	"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
 )
 
+// ratchetLabel domain-separates the expiry ratchet from other uses of the
+// rosenpass output key.
+const ratchetLabel = "netbird-rosenpass-expiry-ratchet"
+
+// RatchetKey derives the successor preshared key from the previous Rosenpass
+// output key. When a key expires without a completed renewal, both peers
+// advance their last shared key by one ratchet step: the expired key is
+// rotated out while both ends still converge on an identical, non-public
+// replacement without communicating.
+func RatchetKey(prev wgtypes.Key) wgtypes.Key {
+	input := make([]byte, 0, len(ratchetLabel)+len(prev))
+	input = append(input, ratchetLabel...)
+	input = append(input, prev[:]...)
+	return sha256.Sum256(input)
+}
+
 // DeterministicSeedKey derives a 32-byte WireGuard preshared key from a pair
 // of peer public keys. Both peers, given the same key pair, produce the same
 // output regardless of which side runs the function: the inputs are ordered

client/internal/routemanager/manager.go

@@ -12,6 +12,7 @@ import (
 	"strings"
 	"sync"
 	"sync/atomic"
+	"syscall"
 	"time"
 
 	"github.com/google/uuid"
@@ -264,7 +265,11 @@ func (m *DefaultManager) initSelector() *routeselector.RouteSelector {
 
 	// restore selector state if it exists
 	if err := m.stateManager.LoadState(state); err != nil {
-		log.Warnf("failed to load state: %v", err)
+		if errors.Is(err, syscall.ENOSYS) {
+			log.Debugf("route selector state unavailable on this platform: %v", err)
+		} else {
+			log.Warnf("failed to load state: %v", err)
+		}
 		return routeselector.NewRouteSelector()
 	}
 

client/ios/NetBirdSDK/env_list.go

@@ -38,7 +38,7 @@ func GetEnvKeyNBForceRelay() string {
 
 // GetEnvKeyNBLazyConn Exports the environment variable for the iOS client
 func GetEnvKeyNBLazyConn() string {
-	return lazyconn.EnvEnableLazyConn
+	return lazyconn.EnvLazyConn
 }
 
 // GetEnvKeyNBInactivityThreshold Exports the environment variable for the iOS client

client/mdm/canonical_loaders.go

@@ -28,6 +28,7 @@ var allKeys = []string{
 	KeyWireguardPort,
 	KeySplitTunnelMode,
 	KeySplitTunnelApps,
+	KeyLazyConnection,
 }
 
 // canonicalKey maps the lowercase form of a managed-config value name to

client/mdm/policy.go

@@ -11,6 +11,7 @@ package mdm
 import (
 	"sort"
 	"strconv"
+	"strings"
 
 	log "github.com/sirupsen/logrus"
 )
@@ -48,6 +49,11 @@ const (
 	// construction — only one mode can be set at a time.
 	KeySplitTunnelMode = "splitTunnelMode"
 	KeySplitTunnelApps = "splitTunnelApps"
+
+	// KeyLazyConnection forces the lazy-connection feature on or off, overriding
+	// the management feature flag. Read as a bool (native bool, or on/off,
+	// true/false, 1/0, yes/no); absent = defer to management.
+	KeyLazyConnection = "lazyConnection"
 )
 
 // Split-tunnel mode literals (KeySplitTunnelMode values).
@@ -69,12 +75,13 @@ var boolStringLiterals = map[string]bool{
 	"true":  true,
 	"1":     true,
 	"yes":   true,
+	"on":    true,
 	"false": false,
 	"0":     false,
 	"no":    false,
+	"off":   false,
 }
 
-
 // Policy holds MDM-managed settings read from the platform source. A nil or
 // empty Policy means no enforcement is active.
 type Policy struct {
@@ -157,7 +164,8 @@ func (p *Policy) GetString(key string) (string, bool) {
 }
 
 // GetBool returns the managed value for key coerced to bool, and whether the
-// key was set. Accepts native bool and string literals "true"/"false"/"1"/"0".
+// key was set. Accepts native bool and string literals (true/false, 1/0,
+// yes/no, on/off), case-insensitively and trimmed of surrounding whitespace.
 func (p *Policy) GetBool(key string) (bool, bool) {
 	if p == nil {
 		return false, false
@@ -170,7 +178,7 @@ func (p *Policy) GetBool(key string) (bool, bool) {
 	case bool:
 		return t, true
 	case string:
-		b, known := boolStringLiterals[t]
+		b, known := boolStringLiterals[strings.ToLower(strings.TrimSpace(t))]
 		return b, known
 	case int:
 		return t != 0, true

client/mdm/policy_test.go

@@ -31,8 +31,8 @@ func TestPolicy_Empty(t *testing.T) {
 
 func TestPolicy_HasKey(t *testing.T) {
 	p := NewPolicy(map[string]any{
-		KeyManagementURL:    "https://corp.example.com",
-		KeyDisableProfiles:  true,
+		KeyManagementURL:   "https://corp.example.com",
+		KeyDisableProfiles: true,
 	})
 	assert.False(t, p.IsEmpty())
 	assert.True(t, p.HasKey(KeyManagementURL))
@@ -53,8 +53,8 @@ func TestPolicy_ManagedKeysSorted(t *testing.T) {
 func TestPolicy_GetString(t *testing.T) {
 	p := NewPolicy(map[string]any{
 		KeyManagementURL:   "https://corp.example.com",
-		KeyDisableProfiles: true,            // wrong type for GetString
-		KeyPreSharedKey:        "",              // empty rejected
+		KeyDisableProfiles: true, // wrong type for GetString
+		KeyPreSharedKey:    "",   // empty rejected
 	})
 	v, ok := p.GetString(KeyManagementURL)
 	assert.True(t, ok)
@@ -85,6 +85,11 @@ func TestPolicy_GetBool(t *testing.T) {
 		{"string 0", "0", false, true},
 		{"string yes", "yes", true, true},
 		{"string no", "no", false, true},
+		{"string on", "on", true, true},
+		{"string off", "off", false, true},
+		{"mixed case On", "On", true, true},
+		{"upper TRUE", "TRUE", true, true},
+		{"padded yes", "  yes  ", true, true},
 		{"int nonzero", 1, true, true},
 		{"int zero", 0, false, true},
 		{"int64 nonzero", int64(2), true, true},

client/proto/daemon_gateway_test.go

@@ -0,0 +1,80 @@
+package proto
+
+import (
+	"context"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	gatewayruntime "github.com/grpc-ecosystem/grpc-gateway/v2/runtime"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/test/bufconn"
+)
+
+func TestGatewayServerRoutesCoverDaemonRPCs(t *testing.T) {
+	mux := gatewayruntime.NewServeMux()
+	if err := RegisterDaemonServiceHandlerServer(context.Background(), mux, UnimplementedDaemonServiceServer{}); err != nil {
+		t.Fatalf("register daemon gateway server handlers: %v", err)
+	}
+
+	assertAllDaemonGatewayRoutesRegistered(t, mux)
+}
+
+func TestGatewayClientRoutesCoverDaemonRPCs(t *testing.T) {
+	listener := bufconn.Listen(1024 * 1024)
+	server := grpc.NewServer()
+	RegisterDaemonServiceServer(server, UnimplementedDaemonServiceServer{})
+	go func() {
+		if err := server.Serve(listener); err != nil && err != grpc.ErrServerStopped {
+			t.Errorf("serve bufconn gRPC server: %v", err)
+		}
+	}()
+	t.Cleanup(func() {
+		server.Stop()
+		_ = listener.Close()
+	})
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	mux := gatewayruntime.NewServeMux()
+	opts := []grpc.DialOption{
+		grpc.WithContextDialer(func(context.Context, string) (net.Conn, error) {
+			return listener.Dial()
+		}),
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+	}
+	if err := RegisterDaemonServiceHandlerFromEndpoint(ctx, mux, "passthrough:///bufnet", opts); err != nil {
+		t.Fatalf("register daemon gateway client handlers: %v", err)
+	}
+
+	assertAllDaemonGatewayRoutesRegistered(t, mux)
+}
+
+func assertAllDaemonGatewayRoutesRegistered(t *testing.T, mux http.Handler) {
+	t.Helper()
+	for _, method := range DaemonService_ServiceDesc.Methods {
+		assertGatewayRouteRegistered(t, mux, method.MethodName)
+	}
+	for _, stream := range DaemonService_ServiceDesc.Streams {
+		assertGatewayRouteRegistered(t, mux, stream.StreamName)
+	}
+}
+
+func assertGatewayRouteRegistered(t *testing.T, mux http.Handler, methodName string) {
+	t.Helper()
+
+	path := "/daemon.DaemonService/" + methodName
+	req := httptest.NewRequest(http.MethodPost, path, strings.NewReader("{}"))
+	req.Header.Set("Content-Type", "application/json")
+	res := httptest.NewRecorder()
+
+	mux.ServeHTTP(res, req)
+
+	if res.Code == http.StatusNotFound {
+		t.Fatalf("gateway route for %s is not registered", methodName)
+	}
+}

client/proto/generate.sh

@@ -12,5 +12,11 @@ script_path=$(dirname "$(realpath "$0")")
 cd "$script_path"
 go install google.golang.org/protobuf/cmd/protoc-gen-go@v1.36.6
 go install google.golang.org/grpc/cmd/protoc-gen-go-grpc@v1.6.1
-protoc -I ./ ./daemon.proto --go_out=../ --go-grpc_out=../ --experimental_allow_proto3_optional
+go install github.com/grpc-ecosystem/grpc-gateway/v2/protoc-gen-grpc-gateway@v2.26.3
+protoc -I ./ ./daemon.proto \
+  --go_out=../ \
+  --go-grpc_out=../ \
+  --grpc-gateway_out=../ \
+  --grpc-gateway_opt=generate_unbound_methods=true \
+  --experimental_allow_proto3_optional
 cd "$old_pwd"

client/server/mdm.go

@@ -3,6 +3,7 @@ package server
 import (
 	"context"
 	"fmt"
+	"net/url"
 	"time"
 
 	log "github.com/sirupsen/logrus"
@@ -155,7 +156,6 @@ func (s *Server) restartEngineForMDMLocked() error {
 	s.config = config
 	s.statusRecorder.UpdateManagementAddress(config.ManagementURL.String())
 	s.statusRecorder.UpdateRosenpass(config.RosenpassEnabled, config.RosenpassPermissive)
-	s.statusRecorder.UpdateLazyConnection(config.LazyConnectionEnabled)
 
 	ctx, cancel := context.WithCancel(s.rootCtx)
 	s.actCancel = cancel
@@ -185,6 +185,37 @@ func conflictBool(key string, p *bool) conflictCheck {
 	}
 }
 
+func canonicalURL(s string) string {
+	u, err := url.ParseRequestURI(s)
+	if err != nil {
+		return s
+	}
+	if u.Port() == "" {
+		switch u.Scheme {
+		case "https":
+			u.Host += ":443"
+		case "http":
+			u.Host += ":80"
+		}
+	}
+	return u.String()
+}
+
+// conflictURL is conflictString for URL-typed keys: both sides are
+// normalized via canonicalURL before comparison.
+func conflictURL(key, got string) conflictCheck {
+	return conflictCheck{
+		key: key,
+		check: func(pol *mdm.Policy) bool {
+			if got == "" {
+				return true
+			}
+			want, ok := pol.GetString(key)
+			return ok && canonicalURL(want) == canonicalURL(got)
+		},
+	}
+}
+
 // conflictString builds a conflictCheck for a string MDM key. An empty
 // `got` is treated as "field not set" (no override requested); otherwise
 // the check returns true only when the policy contains the key and its
@@ -260,7 +291,7 @@ func mdmManagedFieldConflicts(msg *proto.SetConfigRequest, policy *mdm.Policy) [
 	}
 
 	return resolveConflicts(policy, []conflictCheck{
-		conflictString(mdm.KeyManagementURL, msg.ManagementUrl),
+		conflictURL(mdm.KeyManagementURL, msg.ManagementUrl),
 		conflictString(mdm.KeyPreSharedKey, pskGot),
 		conflictBool(mdm.KeyRosenpassEnabled, msg.RosenpassEnabled),
 		conflictBool(mdm.KeyRosenpassPermissive, msg.RosenpassPermissive),
@@ -308,7 +339,6 @@ func setConfigRequestHasConfigOverrides(msg *proto.SetConfigRequest) bool {
 		msg.DisableFirewall != nil ||
 		msg.BlockLanAccess != nil ||
 		msg.DisableNotifications != nil ||
-		msg.LazyConnectionEnabled != nil ||
 		msg.BlockInbound != nil ||
 		msg.DisableIpv6 != nil ||
 		msg.EnableSSHRoot != nil ||
@@ -351,7 +381,6 @@ func loginRequestHasConfigOverrides(msg *proto.LoginRequest) bool {
 		msg.BlockLanAccess != nil ||
 		msg.DisableNotifications != nil ||
 		len(msg.DnsLabels) > 0 || msg.CleanDNSLabels ||
-		msg.LazyConnectionEnabled != nil ||
 		msg.BlockInbound != nil
 }
 
@@ -383,7 +412,7 @@ func loginRequestMDMConflicts(msg *proto.LoginRequest, policy *mdm.Policy) []str
 	}
 
 	return resolveConflicts(policy, []conflictCheck{
-		conflictString(mdm.KeyManagementURL, msg.ManagementUrl),
+		conflictURL(mdm.KeyManagementURL, msg.ManagementUrl),
 		conflictString(mdm.KeyPreSharedKey, pskGot),
 		conflictBool(mdm.KeyRosenpassEnabled, msg.RosenpassEnabled),
 		conflictBool(mdm.KeyRosenpassPermissive, msg.RosenpassPermissive),

client/server/server.go

@@ -238,7 +238,6 @@ func (s *Server) Start() error {
 
 	s.statusRecorder.UpdateManagementAddress(config.ManagementURL.String())
 	s.statusRecorder.UpdateRosenpass(config.RosenpassEnabled, config.RosenpassPermissive)
-	s.statusRecorder.UpdateLazyConnection(config.LazyConnectionEnabled)
 
 	if s.sessionWatcher == nil {
 		s.sessionWatcher = internal.NewSessionWatcher(s.rootCtx, s.statusRecorder)
@@ -422,16 +421,6 @@ func (s *Server) SetConfig(callerCtx context.Context, msg *proto.SetConfigReques
 		return nil, fmt.Errorf("failed to update profile config: %w", err)
 	}
 
-	// Apply the lazy connection toggle to the running engine so it takes
-	// effect without a down/up. s.mutex is already held.
-	if msg.LazyConnectionEnabled != nil && s.connectClient != nil {
-		if engine := s.connectClient.Engine(); engine != nil {
-			if err := engine.SetLazyConnEnabled(msg.GetLazyConnectionEnabled()); err != nil {
-				log.Errorf("failed to apply lazy connection change at runtime: %v", err)
-			}
-		}
-	}
-
 	return &proto.SetConfigResponse{}, nil
 }
 
@@ -510,7 +499,6 @@ func (s *Server) setConfigInputFromRequest(msg *proto.SetConfigRequest) (profile
 	config.DisableFirewall = msg.DisableFirewall
 	config.BlockLANAccess = msg.BlockLanAccess
 	config.DisableNotifications = msg.DisableNotifications
-	config.LazyConnectionEnabled = msg.LazyConnectionEnabled
 	config.BlockInbound = msg.BlockInbound
 	config.DisableIPv6 = msg.DisableIpv6
 	config.EnableSSHRoot = msg.EnableSSHRoot
@@ -840,6 +828,7 @@ func (s *Server) WaitSSOLogin(callerCtx context.Context, msg *proto.WaitSSOLogin
 		return nil, err
 	}
 
+	log.Infof("SSO login flow finished, returning success to caller")
 	return &proto.WaitSSOLoginResponse{
 		Email: tokenInfo.Email,
 	}, nil
@@ -847,6 +836,7 @@ func (s *Server) WaitSSOLogin(callerCtx context.Context, msg *proto.WaitSSOLogin
 
 // Up starts engine work in the daemon.
 func (s *Server) Up(callerCtx context.Context, msg *proto.UpRequest) (*proto.UpResponse, error) {
+	log.Infof("up request received")
 	s.mutex.Lock()
 	// clientRunning is the daemon-intent flag (set by previous Up/Start, cleared
 	// by Down). connectionGoroutineRunning() reports whether the previous retry-loop
@@ -1981,7 +1971,6 @@ func (s *Server) GetConfig(ctx context.Context, req *proto.GetConfigRequest) (*p
 		ServerSSHAllowed:              *cfg.ServerSSHAllowed,
 		RosenpassEnabled:              cfg.RosenpassEnabled,
 		RosenpassPermissive:           cfg.RosenpassPermissive,
-		LazyConnectionEnabled:         cfg.LazyConnectionEnabled,
 		BlockInbound:                  cfg.BlockInbound,
 		DisableNotifications:          disableNotifications,
 		NetworkMonitor:                networkMonitor,

client/server/setconfig_mdm_test.go

@@ -181,6 +181,43 @@ func TestSetConfig_MDMAllow_NonManagedFields(t *testing.T) {
 	require.NotNil(t, resp)
 }
 
+// TestSetConfig_MDMAllow_ManagementURLPortNormalized covers the
+// regression from discussion #6483: MDM URL without explicit port vs
+// UI echo with the parseURL-appended default port must be treated as
+// a no-op echo, not a conflict.
+func TestSetConfig_MDMAllow_ManagementURLPortNormalized(t *testing.T) {
+	tests := []struct {
+		name      string
+		mdmURL    string
+		submitURL string
+	}{
+		{"policy_no_port_submit_with_443", "https://netbird.corp.example", "https://netbird.corp.example:443"},
+		{"policy_with_443_submit_no_port", "https://netbird.corp.example:443", "https://netbird.corp.example"},
+		{"http_policy_no_port_submit_with_80", "http://netbird.corp.example", "http://netbird.corp.example:80"},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			withMDMPolicy(t, mdm.NewPolicy(map[string]any{
+				mdm.KeyManagementURL: tc.mdmURL,
+			}))
+
+			s, ctx, profName, username, _ := setupServerWithProfile(t)
+
+			rosenpassEnabled := true
+			resp, err := s.SetConfig(ctx, &proto.SetConfigRequest{
+				ProfileName:      profName,
+				Username:         username,
+				ManagementUrl:    tc.submitURL,
+				RosenpassEnabled: &rosenpassEnabled,
+			})
+
+			require.NoError(t, err, "port-normalized URL echo must not trip MDM conflict gate")
+			require.NotNil(t, resp)
+		})
+	}
+}
+
 func TestSetConfig_MDMEmpty_NoEnforcement(t *testing.T) {
 	// No MDM policy active: any field can be written.
 	withMDMPolicy(t, mdm.NewPolicy(nil))

client/server/setconfig_test.go

@@ -69,43 +69,41 @@ func TestSetConfig_AllFieldsSaved(t *testing.T) {
 	disableFirewall := true
 	blockLANAccess := true
 	disableNotifications := true
-	lazyConnectionEnabled := true
 	blockInbound := true
 	disableIPv6 := true
 	mtu := int64(1280)
 	sshJWTCacheTTL := int32(300)
 
 	req := &proto.SetConfigRequest{
-		ProfileName:           profName,
-		Username:              currUser.Username,
-		ManagementUrl:         "https://new-api.netbird.io:443",
-		AdminURL:              "https://new-admin.netbird.io",
-		RosenpassEnabled:      &rosenpassEnabled,
-		RosenpassPermissive:   &rosenpassPermissive,
-		ServerSSHAllowed:      &serverSSHAllowed,
-		InterfaceName:         &interfaceName,
-		WireguardPort:         &wireguardPort,
-		OptionalPreSharedKey:  &preSharedKey,
-		DisableAutoConnect:    &disableAutoConnect,
-		NetworkMonitor:        &networkMonitor,
-		DisableClientRoutes:   &disableClientRoutes,
-		DisableServerRoutes:   &disableServerRoutes,
-		DisableDns:            &disableDNS,
-		DisableFirewall:       &disableFirewall,
-		BlockLanAccess:        &blockLANAccess,
-		DisableNotifications:  &disableNotifications,
-		LazyConnectionEnabled: &lazyConnectionEnabled,
-		BlockInbound:          &blockInbound,
-		DisableIpv6:           &disableIPv6,
-		NatExternalIPs:        []string{"1.2.3.4", "5.6.7.8"},
-		CleanNATExternalIPs:   false,
-		CustomDNSAddress:      []byte("1.1.1.1:53"),
-		ExtraIFaceBlacklist:   []string{"eth1", "eth2"},
-		DnsLabels:             []string{"label1", "label2"},
-		CleanDNSLabels:        false,
-		DnsRouteInterval:      durationpb.New(2 * time.Minute),
-		Mtu:                   &mtu,
-		SshJWTCacheTTL:        &sshJWTCacheTTL,
+		ProfileName:          profName,
+		Username:             currUser.Username,
+		ManagementUrl:        "https://new-api.netbird.io:443",
+		AdminURL:             "https://new-admin.netbird.io",
+		RosenpassEnabled:     &rosenpassEnabled,
+		RosenpassPermissive:  &rosenpassPermissive,
+		ServerSSHAllowed:     &serverSSHAllowed,
+		InterfaceName:        &interfaceName,
+		WireguardPort:        &wireguardPort,
+		OptionalPreSharedKey: &preSharedKey,
+		DisableAutoConnect:   &disableAutoConnect,
+		NetworkMonitor:       &networkMonitor,
+		DisableClientRoutes:  &disableClientRoutes,
+		DisableServerRoutes:  &disableServerRoutes,
+		DisableDns:           &disableDNS,
+		DisableFirewall:      &disableFirewall,
+		BlockLanAccess:       &blockLANAccess,
+		DisableNotifications: &disableNotifications,
+		BlockInbound:         &blockInbound,
+		DisableIpv6:          &disableIPv6,
+		NatExternalIPs:       []string{"1.2.3.4", "5.6.7.8"},
+		CleanNATExternalIPs:  false,
+		CustomDNSAddress:     []byte("1.1.1.1:53"),
+		ExtraIFaceBlacklist:  []string{"eth1", "eth2"},
+		DnsLabels:            []string{"label1", "label2"},
+		CleanDNSLabels:       false,
+		DnsRouteInterval:     durationpb.New(2 * time.Minute),
+		Mtu:                  &mtu,
+		SshJWTCacheTTL:       &sshJWTCacheTTL,
 	}
 
 	_, err = s.SetConfig(ctx, req)
@@ -140,7 +138,6 @@ func TestSetConfig_AllFieldsSaved(t *testing.T) {
 	require.Equal(t, blockLANAccess, cfg.BlockLANAccess)
 	require.NotNil(t, cfg.DisableNotifications)
 	require.Equal(t, disableNotifications, *cfg.DisableNotifications)
-	require.Equal(t, lazyConnectionEnabled, cfg.LazyConnectionEnabled)
 	require.Equal(t, blockInbound, cfg.BlockInbound)
 	require.Equal(t, disableIPv6, cfg.DisableIPv6)
 	require.Equal(t, []string{"1.2.3.4", "5.6.7.8"}, cfg.NATExternalIPs)
@@ -164,13 +161,14 @@ func verifyAllFieldsCovered(t *testing.T, req *proto.SetConfigRequest) {
 	t.Helper()
 
 	metadataFields := map[string]bool{
-		"state":               true, // protobuf internal
-		"sizeCache":           true, // protobuf internal
-		"unknownFields":       true, // protobuf internal
-		"Username":            true, // metadata
-		"ProfileName":         true, // metadata
-		"CleanNATExternalIPs": true, // control flag for clearing
-		"CleanDNSLabels":      true, // control flag for clearing
+		"state":                 true, // protobuf internal
+		"sizeCache":             true, // protobuf internal
+		"unknownFields":         true, // protobuf internal
+		"Username":              true, // metadata
+		"ProfileName":           true, // metadata
+		"CleanNATExternalIPs":   true, // control flag for clearing
+		"CleanDNSLabels":        true, // control flag for clearing
+		"LazyConnectionEnabled": true, // deprecated: proto field retained for compat, no longer applied
 	}
 
 	expectedFields := map[string]bool{
@@ -190,7 +188,6 @@ func verifyAllFieldsCovered(t *testing.T, req *proto.SetConfigRequest) {
 		"DisableFirewall":               true,
 		"BlockLanAccess":                true,
 		"DisableNotifications":          true,
-		"LazyConnectionEnabled":         true,
 		"BlockInbound":                  true,
 		"DisableIpv6":                   true,
 		"NatExternalIPs":                true,
@@ -252,7 +249,6 @@ func TestCLIFlags_MappedToSetConfig(t *testing.T) {
 		"block-lan-access":                  "BlockLanAccess",
 		"block-inbound":                     "BlockInbound",
 		"disable-ipv6":                      "DisableIpv6",
-		"enable-lazy-connection":            "LazyConnectionEnabled",
 		"external-ip-map":                   "NatExternalIPs",
 		"dns-resolver-address":              "CustomDNSAddress",
 		"extra-iface-blacklist":             "ExtraIFaceBlacklist",
@@ -269,7 +265,8 @@ func TestCLIFlags_MappedToSetConfig(t *testing.T) {
 
 	// SetConfigRequest fields that don't have CLI flags (settable only via UI or other means).
 	fieldsWithoutCLIFlags := map[string]bool{
-		"DisableNotifications": true, // Only settable via UI
+		"DisableNotifications":  true, // Only settable via UI
+		"LazyConnectionEnabled": true, // deprecated: no longer settable (managed by server + NB_LAZY_CONN)
 	}
 
 	// Get all SetConfigRequest fields to verify our map is complete.

client/system/info.go

@@ -74,8 +74,6 @@ type Info struct {
 	BlockInbound        bool
 	DisableIPv6         bool
 
-	LazyConnectionEnabled bool
-
 	EnableSSHRoot                 bool
 	EnableSSHSFTP                 bool
 	EnableSSHLocalPortForwarding  bool
@@ -87,7 +85,7 @@ func (i *Info) SetFlags(
 	rosenpassEnabled, rosenpassPermissive bool,
 	serverSSHAllowed *bool,
 	disableClientRoutes, disableServerRoutes,
-	disableDNS, disableFirewall, blockLANAccess, blockInbound, disableIPv6, lazyConnectionEnabled bool,
+	disableDNS, disableFirewall, blockLANAccess, blockInbound, disableIPv6 bool,
 	enableSSHRoot, enableSSHSFTP, enableSSHLocalPortForwarding, enableSSHRemotePortForwarding *bool,
 	disableSSHAuth *bool,
 ) {
@@ -105,8 +103,6 @@ func (i *Info) SetFlags(
 	i.BlockInbound = blockInbound
 	i.DisableIPv6 = disableIPv6
 
-	i.LazyConnectionEnabled = lazyConnectionEnabled
-
 	if enableSSHRoot != nil {
 		i.EnableSSHRoot = *enableSSHRoot
 	}

client/system/process.go

@@ -7,7 +7,7 @@ import (
 	"os"
 	"slices"
 
-	"github.com/shirou/gopsutil/v3/process"
+	"github.com/shirou/gopsutil/v4/process"
 )
 
 // getRunningProcesses returns a list of running process paths. The context bounds the work:

client/system/process_test.go

@@ -4,7 +4,7 @@ import (
 	"context"
 	"testing"
 
-	"github.com/shirou/gopsutil/v3/process"
+	"github.com/shirou/gopsutil/v4/process"
 )
 
 func Benchmark_getRunningProcesses(b *testing.B) {

client/test/json-socket-docker.sh

@@ -0,0 +1,223 @@
+#!/usr/bin/env bash
+set -eEuo pipefail
+
+usage() {
+  cat <<'EOF'
+Usage: client/test/json-socket-docker.sh [tcp|unix|both]
+
+Builds the NetBird client Docker image from the local source tree, starts
+`netbird service run` in a container with --enable-json-socket, and verifies
+that the HTTP/JSON daemon gateway responds to Status requests.
+
+Modes:
+  tcp   Validate tcp://0.0.0.0:8080 via a published localhost port (default)
+  unix  Validate unix:///sock/netbird-http.sock via a bind-mounted socket dir
+  both  Run both validations
+
+Environment:
+  CONTAINER_RUNTIME  docker or podman. Auto-detected if unset.
+  IMAGE              Image tag to build. Default: netbird-json-socket-test:local
+  TARGETARCH         Go/Docker target arch. Default: `go env GOARCH`
+  PLATFORM           Docker platform. Default: linux/$TARGETARCH
+  WAIT_TIMEOUT       Seconds to wait for the JSON socket. Default: 30
+EOF
+}
+
+if [[ "${1:-}" == "-h" || "${1:-}" == "--help" ]]; then
+  usage
+  exit 0
+fi
+
+MODE="${1:-tcp}"
+case "${MODE}" in
+  tcp|unix|both) ;;
+  *)
+    usage >&2
+    echo "invalid mode: ${MODE}" >&2
+    exit 2
+    ;;
+esac
+
+RUNTIME="${CONTAINER_RUNTIME:-}"
+if [[ -z "${RUNTIME}" ]]; then
+  if command -v docker >/dev/null 2>&1; then
+    RUNTIME=docker
+  elif command -v podman >/dev/null 2>&1; then
+    RUNTIME=podman
+  else
+    echo "docker or podman is required" >&2
+    exit 127
+  fi
+fi
+if ! command -v "${RUNTIME}" >/dev/null 2>&1; then
+  echo "container runtime not found: ${RUNTIME}" >&2
+  exit 127
+fi
+
+if ! command -v curl >/dev/null 2>&1; then
+  echo "curl is required" >&2
+  exit 127
+fi
+
+ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)"
+IMAGE="${IMAGE:-netbird-json-socket-test:local}"
+TARGETARCH="${TARGETARCH:-$(go env GOARCH)}"
+PLATFORM="${PLATFORM:-linux/${TARGETARCH}}"
+WAIT_TIMEOUT="${WAIT_TIMEOUT:-30}"
+TMP_DIR="$(mktemp -d)"
+CONTAINERS=()
+
+cleanup() {
+  local status=$?
+  for container in "${CONTAINERS[@]:-}"; do
+    "${RUNTIME}" rm -f "${container}" >/dev/null 2>&1 || true
+  done
+  rm -rf "${TMP_DIR}"
+  exit "${status}"
+}
+trap cleanup EXIT
+
+build_image() {
+  echo "==> Building Linux ${TARGETARCH} netbird binary"
+  mkdir -p "${TMP_DIR}/context/client"
+  cp "${ROOT_DIR}/client/Dockerfile" "${TMP_DIR}/context/Dockerfile"
+  cp "${ROOT_DIR}/client/netbird-entrypoint.sh" "${TMP_DIR}/context/client/netbird-entrypoint.sh"
+
+  (cd "${ROOT_DIR}" && CGO_ENABLED=0 GOOS=linux GOARCH="${TARGETARCH}" go build -o "${TMP_DIR}/context/netbird" ./client)
+
+  echo "==> Building ${IMAGE} for ${PLATFORM}"
+  "${RUNTIME}" build \
+    --platform "${PLATFORM}" \
+    --build-arg NETBIRD_BINARY=netbird \
+    -t "${IMAGE}" \
+    -f "${TMP_DIR}/context/Dockerfile" \
+    "${TMP_DIR}/context"
+}
+
+pick_port() {
+  python3 - <<'PY'
+import socket
+sock = socket.socket()
+sock.bind(("127.0.0.1", 0))
+print(sock.getsockname()[1])
+sock.close()
+PY
+}
+
+assert_status_json() {
+  local response_file="$1"
+  if command -v python3 >/dev/null 2>&1; then
+    python3 - "${response_file}" <<'PY'
+import json
+import sys
+with open(sys.argv[1], encoding="utf-8") as fh:
+    data = json.load(fh)
+if not data.get("status"):
+    raise SystemExit("missing non-empty status field")
+if "daemonVersion" not in data:
+    raise SystemExit("missing daemonVersion field")
+print(f"status={data['status']} daemonVersion={data['daemonVersion']}")
+PY
+  else
+    grep -q '"status"' "${response_file}"
+    grep -q '"daemonVersion"' "${response_file}"
+    cat "${response_file}"
+  fi
+}
+
+container_logs() {
+  local container="$1"
+  echo "---- ${container} logs ----" >&2
+  "${RUNTIME}" logs "${container}" >&2 || true
+  echo "--------------------------" >&2
+}
+
+wait_for_http_status() {
+  local container="$1"
+  local response="${TMP_DIR}/${container}.json"
+  local curl_err="${TMP_DIR}/${container}.curl.err"
+  shift
+  local deadline=$((SECONDS + WAIT_TIMEOUT))
+
+  while (( SECONDS < deadline )); do
+    if curl -fsS "$@" \
+      -X POST \
+      -H 'Content-Type: application/json' \
+      -d '{}' \
+      -o "${response}" \
+      2>"${curl_err}"; then
+      assert_status_json "${response}"
+      return 0
+    fi
+
+    if ! "${RUNTIME}" ps --format '{{.Names}}' | grep -Fxq "${container}"; then
+      echo "container exited before JSON socket became ready" >&2
+      container_logs "${container}"
+      return 1
+    fi
+    sleep 1
+  done
+
+  echo "timed out waiting for JSON socket after ${WAIT_TIMEOUT}s" >&2
+  cat "${curl_err}" >&2 || true
+  container_logs "${container}"
+  return 1
+}
+
+run_netbird_container() {
+  local container="$1"
+  local json_socket="$2"
+  shift 2
+
+  CONTAINERS+=("${container}")
+  "${RUNTIME}" run --rm -d \
+    --name "${container}" \
+    -e NB_STATE_DIR=/tmp/netbird-state \
+    --entrypoint /usr/local/bin/netbird \
+    "$@" \
+    "${IMAGE}" \
+    --log-file console \
+    --daemon-addr unix:///tmp/netbird.sock \
+    service run \
+    --enable-json-socket \
+    --json-socket "${json_socket}" >/dev/null
+}
+
+run_tcp_test() {
+  local port container
+  port="$(pick_port)"
+  container="nb-json-socket-tcp-$RANDOM-$RANDOM"
+
+  echo "==> Validating TCP JSON socket on 127.0.0.1:${port}"
+  run_netbird_container "${container}" "tcp://0.0.0.0:8080" -p "127.0.0.1:${port}:8080"
+  wait_for_http_status "${container}" "http://127.0.0.1:${port}/daemon.DaemonService/Status"
+}
+
+run_unix_test() {
+  local sock_dir sock_path container
+  sock_dir="${TMP_DIR}/sock"
+  sock_path="${sock_dir}/netbird-http.sock"
+  container="nb-json-socket-unix-$RANDOM-$RANDOM"
+  mkdir -p "${sock_dir}"
+
+  echo "==> Validating Unix JSON socket at ${sock_path}"
+  run_netbird_container "${container}" "unix:///sock/netbird-http.sock" -v "${sock_dir}:/sock"
+  wait_for_http_status "${container}" --unix-socket "${sock_path}" "http://unix/daemon.DaemonService/Status"
+}
+
+build_image
+
+case "${MODE}" in
+  tcp)
+    run_tcp_test
+    ;;
+  unix)
+    run_unix_test
+    ;;
+  both)
+    run_tcp_test
+    run_unix_test
+    ;;
+esac
+
+echo "==> Docker JSON socket validation passed (${MODE})"