fix linter

fix interval and job cancel
improve jobs endpoint
2026-07-22 08:21:30 +02:00 · 2026-07-16 13:24:34 +02:00 · 2026-07-16 02:13:51 +02:00 · 2026-07-16 02:07:15 +02:00 · 2026-07-16 01:22:37 +02:00 · 2026-07-16 01:22:30 +02:00
37 changed files with 1880 additions and 1260 deletions
--- a/client/cmd/login.go
+++ b/client/cmd/login.go
@@ -17,7 +17,9 @@ import (
 	"github.com/netbirdio/netbird/client/internal"
 	"github.com/netbirdio/netbird/client/internal/auth"
 	"github.com/netbirdio/netbird/client/internal/profilemanager"
+	nbnet "github.com/netbirdio/netbird/client/net"
 	"github.com/netbirdio/netbird/client/proto"
+	"github.com/netbirdio/netbird/client/server"
 	"github.com/netbirdio/netbird/client/system"
 	"github.com/netbirdio/netbird/util"
 )
@@ -331,6 +333,14 @@ func doForegroundLogin(ctx context.Context, cmd *cobra.Command, setupKey string,
 		return fmt.Errorf("read config file %s: %v", configFilePath, err)
 	}

+	// Mirror runInForegroundMode: recover residual state (DNS, firewall,
+	// ssh config, legacy routing) from a previous unclean shutdown and
+	// enable advanced routing before dialing management.
+	if err := server.RestoreResidualState(ctx, profilemanager.NewServiceManager(configFilePath).GetStatePath()); err != nil {
+		log.Warnf("failed to restore residual state: %v", err)
+	}
+	nbnet.Init()
+
 	err = foregroundLogin(ctx, cmd, config, setupKey, activeProf.ID)
 	if err != nil {
 		return fmt.Errorf("foreground login failed: %v", err)
--- a/client/cmd/up.go
+++ b/client/cmd/up.go
@@ -22,6 +22,8 @@ import (
 	"github.com/netbirdio/netbird/client/internal/peer"
 	"github.com/netbirdio/netbird/client/internal/profilemanager"
 	"github.com/netbirdio/netbird/client/proto"
+	nbnet "github.com/netbirdio/netbird/client/net"
+	"github.com/netbirdio/netbird/client/server"
 	"github.com/netbirdio/netbird/client/system"
 	"github.com/netbirdio/netbird/shared/management/domain"
 	"github.com/netbirdio/netbird/util"
@@ -229,6 +231,24 @@ func runInForegroundMode(ctx context.Context, cmd *cobra.Command, activeProf *pr

 	_, _ = profilemanager.UpdateOldManagementURL(ctx, config, configFilePath)

+	// Restore residual state left by a previous run that did not shut down
+	// cleanly, mirroring what the daemon does before connecting: it recovers
+	// DNS config (a stale resolv.conf takeover can make the management
+	// hostname unresolvable), firewall rules, ssh config and legacy routing.
+	// Route cleanup itself happens at engine start; nbnet.Init() below lets
+	// the management dial bypass a leftover fwmark rule until then.
+	// Foreground mode is particularly exposed in containers: a crashed
+	// container restarts inside the same (pod) network namespace, so stale
+	// state survives while the process does not.
+	if err := server.RestoreResidualState(ctx, profilemanager.NewServiceManager(configPath).GetStatePath()); err != nil {
+		log.Warnf("failed to restore residual state: %v", err)
+	}
+
+	// Enable advanced routing (as the daemon does on startup) so the
+	// management dial bypasses a leftover fwmark rule instead of being
+	// shunted into a stale routing table.
+	nbnet.Init()
+
 	err = foregroundLogin(ctx, cmd, config, providedSetupKey, activeProf.ID)
 	if err != nil {
 		return fmt.Errorf("foreground login failed: %v", err)
--- a/client/embed/embed.go
+++ b/client/embed/embed.go
@@ -85,11 +85,6 @@ type Options struct {
 	DisableIPv6 bool
 	// BlockInbound blocks all inbound connections from peers
 	BlockInbound bool
-	// EnableRosenpass enables the Rosenpass post-quantum key exchange.
-	EnableRosenpass bool
-	// RosenpassPermissive lets a Rosenpass-enabled peer still connect to peers
-	// that do not run Rosenpass (falling back to the plain WireGuard PSK).
-	RosenpassPermissive bool
 	// BlockLANAccess blocks the embedded peer from reaching the host's
 	// LAN (RFC 1918, link-local, loopback) when it's used as a routing
 	// peer. Mirrors profilemanager.ConfigInput.BlockLANAccess. Useful
@@ -208,8 +203,6 @@ func New(opts Options) (*Client, error) {
 		DisableIPv6:         &opts.DisableIPv6,
 		BlockInbound:        &opts.BlockInbound,
 		BlockLANAccess:      &opts.BlockLANAccess,
-		RosenpassEnabled:    &opts.EnableRosenpass,
-		RosenpassPermissive: &opts.RosenpassPermissive,
 		WireguardPort:       opts.WireguardPort,
 		MTU:                 opts.MTU,
 		DNSLabels:           parsedLabels,
--- a/client/internal/conn_mgr.go
+++ b/client/internal/conn_mgr.go
@@ -2,7 +2,6 @@ package internal

 import (
 	"context"
-	"maps"
 	"os"
 	"strconv"
 	"sync"
@@ -15,7 +14,6 @@ import (
 	"github.com/netbirdio/netbird/client/internal/peer"
 	"github.com/netbirdio/netbird/client/internal/peerstore"
 	"github.com/netbirdio/netbird/route"
-	mgmProto "github.com/netbirdio/netbird/shared/management/proto"
 )

 // lazyForce is the resolved local decision for lazy connections, layered above the
@@ -37,18 +35,13 @@ const (
 //
 // The implementation is not thread-safe; it is protected by engine.syncMsgMux.
 type ConnMgr struct {
-	peerStore      *peerstore.Store
-	statusRecorder *peer.Status
-	iface          lazyconn.WGIface
-	force          lazyForce
-	// remoteLazyEnabled caches the account-wide lazy feature flag from management.
-	// It is the default for peers that do not carry a per-peer lazy hint.
-	remoteLazyEnabled bool
+	peerStore        *peerstore.Store
+	statusRecorder   *peer.Status
+	iface            lazyconn.WGIface
+	force            lazyForce
+	rosenpassEnabled bool

 	lazyConnMgr *manager.Manager
-	// appliedExcludeList is the exclude set last handed to the lazy manager, kept so an
-	// unchanged set on the next sync skips the O(n) reconciliation.
-	appliedExcludeList map[string]bool

 	wg            sync.WaitGroup
 	lazyCtx       context.Context
@@ -57,59 +50,78 @@ type ConnMgr struct {

 func NewConnMgr(engineConfig *EngineConfig, statusRecorder *peer.Status, peerStore *peerstore.Store, iface lazyconn.WGIface) *ConnMgr {
 	e := &ConnMgr{
-		peerStore:      peerStore,
-		statusRecorder: statusRecorder,
-		iface:          iface,
-		force:          resolveLazyForce(engineConfig.LazyConnection),
+		peerStore:        peerStore,
+		statusRecorder:   statusRecorder,
+		iface:            iface,
+		force:            resolveLazyForce(engineConfig.LazyConnection),
+		rosenpassEnabled: engineConfig.RosenpassEnabled,
 	}
 	return e
 }

-// Start initializes the connection manager. The lazy connection manager always runs so that
-// per-peer lazy defaults (e.g. proxy peers) work even when the account flag is off; the
-// account flag and the local override decide the default lazy state per peer (see
-// PeerLazyDefault). Rosenpass peers stay lazy-capable too: their connections just never idle
-// on their own, since rosenpass rekey traffic keeps them active.
+// 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
 	}

-	e.initLazyManager(ctx)
-	e.statusRecorder.UpdateLazyConnection(e.PeerLazyDefault(mgmProto.LazyState_LazyStateDefault))
-}
-
-// UpdatedRemoteFeatureFlag caches the account-wide lazy feature flag. The manager itself is
-// not started or stopped here; the per-sync exclude-list reconciliation moves normal peers
-// between the lazy and always-active sets when the flag flips.
-func (e *ConnMgr) UpdatedRemoteFeatureFlag(_ context.Context, enabled bool) error {
-	e.remoteLazyEnabled = enabled
-	if e.isStartedWithLazyMgr() {
-		e.statusRecorder.UpdateLazyConnection(e.PeerLazyDefault(mgmProto.LazyState_LazyStateDefault))
-	}
-	return nil
-}
-
-// PeerLazyDefault reports whether a peer should be lazy. The local override
-// (NB_LAZY_CONN/MDM) wins over everything; without a local override the
-// management per-peer state applies (LazyStateLazy/Eager force the decision),
-// and LazyStateDefault follows the account-wide flag.
-func (e *ConnMgr) PeerLazyDefault(state mgmProto.LazyState) bool {
 	switch e.force {
-	case lazyForceOn:
-		return true
 	case lazyForceOff:
-		return false
+		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
 	}

-	switch state {
-	case mgmProto.LazyState_LazyStateLazy:
-		return true
-	case mgmProto.LazyState_LazyStateEager:
-		return false
-	default:
-		return e.remoteLazyEnabled
+	if e.rosenpassEnabled {
+		log.Warnf("rosenpass connection manager is enabled, lazy connection manager will not be started")
+		e.statusRecorder.UpdateLazyConnection(false)
+		return
+	}
+
+	e.initLazyManager(ctx)
+	e.statusRecorder.UpdateLazyConnection(true)
+}
+
+// UpdatedRemoteFeatureFlag is called when the remote feature flag is updated.
+// 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 {
+	// a local override (NB_LAZY_CONN or local config) takes precedence over management
+	if e.force != lazyForceNone {
+		return nil
+	}
+
+	if enabled {
+		// if the lazy connection manager is already started, do not start it again
+		if e.lazyConnMgr != nil {
+			return nil
+		}
+
+		if e.rosenpassEnabled {
+			log.Infof("rosenpass connection manager is enabled, lazy connection manager will not be started")
+			e.statusRecorder.UpdateLazyConnection(false)
+			return nil
+		}
+
+		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")
+		e.closeManager(ctx)
+		e.statusRecorder.UpdateLazyConnection(false)
+		return nil
 	}
 }

@@ -129,13 +141,6 @@ func (e *ConnMgr) SetExcludeList(ctx context.Context, peerIDs map[string]bool) {
 		return
 	}

-	// The exclude set is recomputed every sync but rarely changes; skip the O(n)
-	// store lookups and reconciliation when it matches what was already applied.
-	if maps.Equal(peerIDs, e.appliedExcludeList) {
-		return
-	}
-	e.appliedExcludeList = maps.Clone(peerIDs)
-
 	excludedPeers := make([]lazyconn.PeerConfig, 0, len(peerIDs))

 	for peerID := range peerIDs {
@@ -171,16 +176,12 @@ func (e *ConnMgr) SetExcludeList(ctx context.Context, peerIDs map[string]bool) {
 	}
 }

-// AddPeerConn registers a peer connection. permanent requests an always-active connection
-// (the peer belongs to the exclude set: a forwarder, or a peer that is not lazy by policy).
-// Non-permanent peers are handed to the lazy manager. The subsequent SetExcludeList call
-// reconciles membership for existing peers across flag flips.
-func (e *ConnMgr) AddPeerConn(ctx context.Context, peerKey string, conn *peer.Conn, permanent bool) (exists bool) {
+func (e *ConnMgr) AddPeerConn(ctx context.Context, peerKey string, conn *peer.Conn) (exists bool) {
 	if success := e.peerStore.AddPeerConn(peerKey, conn); !success {
 		return true
 	}

-	if !e.isStartedWithLazyMgr() || permanent {
+	if !e.isStartedWithLazyMgr() {
 		if err := conn.Open(ctx); err != nil {
 			conn.Log.Errorf("failed to open connection: %v", err)
 		}
@@ -268,7 +269,6 @@ func (e *ConnMgr) Close() {
 	e.lazyCtxCancel()
 	e.wg.Wait()
 	e.lazyConnMgr = nil
-	e.appliedExcludeList = nil
 }

 func (e *ConnMgr) initLazyManager(engineCtx context.Context) {
@@ -276,7 +276,6 @@ func (e *ConnMgr) initLazyManager(engineCtx context.Context) {
 		InactivityThreshold: inactivityThresholdEnv(),
 	}
 	e.lazyConnMgr = manager.NewManager(cfg, engineCtx, e.peerStore, e.iface)
-	e.appliedExcludeList = nil

 	e.lazyCtx, e.lazyCtxCancel = context.WithCancel(engineCtx)

@@ -287,6 +286,43 @@ func (e *ConnMgr) initLazyManager(engineCtx context.Context) {
 	}()
 }

+func (e *ConnMgr) addPeersToLazyConnManager() error {
+	peers := e.peerStore.PeersPubKey()
+	lazyPeerCfgs := make([]lazyconn.PeerConfig, 0, len(peers))
+	for _, peerID := range peers {
+		var peerConn *peer.Conn
+		var exists bool
+		if peerConn, exists = e.peerStore.PeerConn(peerID); !exists {
+			log.Warnf("failed to find peer conn for peerID: %s", peerID)
+			continue
+		}
+
+		lazyPeerCfg := lazyconn.PeerConfig{
+			PublicKey:  peerID,
+			AllowedIPs: peerConn.WgConfig().AllowedIps,
+			PeerConnID: peerConn.ConnID(),
+			Log:        peerConn.Log,
+		}
+		lazyPeerCfgs = append(lazyPeerCfgs, lazyPeerCfg)
+	}
+
+	return e.lazyConnMgr.AddActivePeers(lazyPeerCfgs)
+}
+
+func (e *ConnMgr) closeManager(ctx context.Context) {
+	if e.lazyConnMgr == nil {
+		return
+	}
+
+	e.lazyCtxCancel()
+	e.wg.Wait()
+	e.lazyConnMgr = nil
+
+	for _, peerID := range e.peerStore.PeersPubKey() {
+		e.peerStore.PeerConnOpen(ctx, peerID)
+	}
+}
+
 func (e *ConnMgr) isStartedWithLazyMgr() bool {
 	return e.lazyConnMgr != nil && e.lazyCtxCancel != nil
 }
--- a/client/internal/conn_mgr_test.go
+++ b/client/internal/conn_mgr_test.go
@@ -5,7 +5,6 @@ import (
 	"testing"

 	"github.com/netbirdio/netbird/client/internal/lazyconn"
-	mgmProto "github.com/netbirdio/netbird/shared/management/proto"
 )

 func TestResolveLazyForce(t *testing.T) {
@@ -39,90 +38,3 @@ func TestResolveLazyForce(t *testing.T) {
 		})
 	}
 }
-
-func TestPeerLazyDefault(t *testing.T) {
-	tests := []struct {
-		name          string
-		force         lazyForce
-		remoteEnabled bool
-		state         mgmProto.LazyState
-		want          bool
-	}{
-		{name: "force on wins over eager state", force: lazyForceOn, state: mgmProto.LazyState_LazyStateEager, want: true},
-		{name: "force off wins over lazy state", force: lazyForceOff, remoteEnabled: true, state: mgmProto.LazyState_LazyStateLazy, want: false},
-		{name: "none, default, account off -> active", force: lazyForceNone, state: mgmProto.LazyState_LazyStateDefault, want: false},
-		{name: "none, default, account on -> lazy", force: lazyForceNone, remoteEnabled: true, state: mgmProto.LazyState_LazyStateDefault, want: true},
-		{name: "none, lazy state, account off -> lazy", force: lazyForceNone, state: mgmProto.LazyState_LazyStateLazy, want: true},
-		{name: "none, eager state, account on -> active", force: lazyForceNone, remoteEnabled: true, state: mgmProto.LazyState_LazyStateEager, want: false},
-	}
-
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			e := &ConnMgr{force: tt.force, remoteLazyEnabled: tt.remoteEnabled}
-			if got := e.PeerLazyDefault(tt.state); got != tt.want {
-				t.Fatalf("PeerLazyDefault(%v) = %v, want %v", tt.state, got, tt.want)
-			}
-		})
-	}
-}
-
-// TestToExcludedLazyPeers covers the per-peer lazy classification (proxy vs
-// normal, across the force/account-flag matrix). Forwarder-target exclusion is
-// covered by TestToExcludedLazyPeers_ForwardTarget.
-func TestToExcludedLazyPeers(t *testing.T) {
-	const (
-		normalKey = "normal"
-		lazyKey   = "lazy-state"
-		eagerKey  = "eager-state"
-	)
-
-	peers := []*mgmProto.RemotePeerConfig{
-		{WgPubKey: normalKey, AllowedIps: []string{"100.64.0.1/32"}},
-		{WgPubKey: lazyKey, AllowedIps: []string{"100.64.0.2/32"}, LazyState: mgmProto.LazyState_LazyStateLazy},
-		{WgPubKey: eagerKey, AllowedIps: []string{"100.64.0.3/32"}, LazyState: mgmProto.LazyState_LazyStateEager},
-	}
-
-	tests := []struct {
-		name          string
-		force         lazyForce
-		remoteEnabled bool
-		want          map[string]bool
-	}{
-		{
-			name:  "account off: lazy-state peer lazy, normal + eager active",
-			force: lazyForceNone, remoteEnabled: false,
-			want: map[string]bool{normalKey: true, eagerKey: true},
-		},
-		{
-			name:  "account on: only eager-state peer active",
-			force: lazyForceNone, remoteEnabled: true,
-			want: map[string]bool{eagerKey: true},
-		},
-		{
-			name:  "force off: everything active",
-			force: lazyForceOff, remoteEnabled: true,
-			want: map[string]bool{normalKey: true, lazyKey: true, eagerKey: true},
-		},
-		{
-			name:  "force on: nothing active",
-			force: lazyForceOn, remoteEnabled: false,
-			want: map[string]bool{},
-		},
-	}
-
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			e := &Engine{connMgr: &ConnMgr{force: tt.force, remoteLazyEnabled: tt.remoteEnabled}}
-			got := e.toExcludedLazyPeers(peers)
-
-			if len(got) != len(tt.want) {
-				t.Fatalf("toExcludedLazyPeers() = %v, want %v", got, tt.want)
-			}
-			for k := range tt.want {
-				if !got[k] {
-					t.Fatalf("expected peer %s excluded, got %v", k, got)
-				}
-			}
-		})
-	}
-}
--- a/client/internal/engine.go
+++ b/client/internal/engine.go
@@ -842,7 +842,8 @@ func (e *Engine) modifyPeers(peersUpdate []*mgmProto.RemotePeerConfig) error {
 	}
 	// third, add the peer connections again
 	for _, p := range modified {
-		if err := e.addNewPeer(p); err != nil {
+		err := e.addNewPeer(p)
+		if err != nil {
 			return err
 		}
 	}
@@ -1402,12 +1403,8 @@ func (e *Engine) updateNetworkMap(networkMap *mgmProto.NetworkMap) error {
 		return nil
 	}

-	// Only update the flag when the sync carries a peer config; a nil peer config
-	// (e.g. a partial update) must not reset the cached flag to false.
-	if peerConfig := networkMap.GetPeerConfig(); peerConfig != nil {
-		if err := e.connMgr.UpdatedRemoteFeatureFlag(e.ctx, peerConfig.GetLazyConnectionEnabled()); err != nil {
-			log.Errorf("failed to update lazy connection feature flag: %v", err)
-		}
+	if err := e.connMgr.UpdatedRemoteFeatureFlag(e.ctx, networkMap.GetPeerConfig().GetLazyConnectionEnabled()); err != nil {
+		log.Errorf("failed to update lazy connection feature flag: %v", err)
 	}

 	if e.firewall != nil {
@@ -1473,7 +1470,8 @@ func (e *Engine) updateNetworkMap(networkMap *mgmProto.NetworkMap) error {

 	// Ingress forward rules
 	done = e.phase("forward_rules")
-	if _, err := e.updateForwardRules(networkMap.GetForwardingRules()); err != nil {
+	forwardingRules, err := e.updateForwardRules(networkMap.GetForwardingRules())
+	if err != nil {
 		log.Errorf("failed to update forward rules, err: %v", err)
 	}
 	done()
@@ -1491,7 +1489,8 @@ func (e *Engine) updateNetworkMap(networkMap *mgmProto.NetworkMap) error {

 	// must set the exclude list after the peers are added. Without it the manager can not figure out the peers parameters from the store
 	done = e.phase("lazy_exclude")
-	e.connMgr.SetExcludeList(e.ctx, e.toExcludedLazyPeers(remotePeers))
+	excludedLazyPeers := e.toExcludedLazyPeers(forwardingRules, remotePeers)
+	e.connMgr.SetExcludeList(e.ctx, excludedLazyPeers)
 	done()

 	e.networkSerial = serial
@@ -1735,15 +1734,15 @@ func addrToString(addr netip.Addr) string {
 // addNewPeers adds peers that were not know before but arrived from the Management service with the update
 func (e *Engine) addNewPeers(peersUpdate []*mgmProto.RemotePeerConfig) error {
 	for _, p := range peersUpdate {
-		if err := e.addNewPeer(p); err != nil {
+		err := e.addNewPeer(p)
+		if err != nil {
 			return err
 		}
 	}
 	return nil
 }

-// addNewPeer add peer if connection doesn't exist. A peer that is not lazy by
-// policy gets an always-active connection instead.
+// addNewPeer add peer if connection doesn't exist
 func (e *Engine) addNewPeer(peerConfig *mgmProto.RemotePeerConfig) error {
 	peerKey := peerConfig.GetWgPubKey()
 	peerIPs := make([]netip.Prefix, 0, len(peerConfig.GetAllowedIps()))
@@ -1778,8 +1777,7 @@ func (e *Engine) addNewPeer(peerConfig *mgmProto.RemotePeerConfig) error {
 		log.Warnf("error adding peer %s to status recorder, got error: %v", peerKey, err)
 	}

-	permanent := !e.connMgr.PeerLazyDefault(peerConfig.GetLazyState())
-	if exists := e.connMgr.AddPeerConn(e.ctx, peerKey, conn, permanent); exists {
+	if exists := e.connMgr.AddPeerConn(e.ctx, peerKey, conn); exists {
 		conn.Close(false)
 		return fmt.Errorf("peer already exists: %s", peerKey)
 	}
@@ -2605,19 +2603,46 @@ func (e *Engine) updateForwardRules(rules []*mgmProto.ForwardingRule) ([]firewal
 	return forwardingRules, nberrors.FormatErrorOrNil(merr)
 }

-// toExcludedLazyPeers returns the peers that must have an always-active
-// connection: those that are not lazy by policy (the per-peer lazy state or the
-// account flag, subject to the local override).
-func (e *Engine) toExcludedLazyPeers(peers []*mgmProto.RemotePeerConfig) map[string]bool {
+func (e *Engine) toExcludedLazyPeers(rules []firewallManager.ForwardRule, peers []*mgmProto.RemotePeerConfig) map[string]bool {
 	excludedPeers := make(map[string]bool)
-	for _, p := range peers {
-		if !e.connMgr.PeerLazyDefault(p.GetLazyState()) {
-			excludedPeers[p.GetWgPubKey()] = true
+
+	// Ingress forward targets: inbound forwarded traffic is initiated remotely and
+	// cannot wake a lazy connection, so the peer routing the target must stay
+	// permanently connected. AllowedIPs are already parsed on the peer conn, so
+	// reuse those typed prefixes instead of re-parsing the network map strings.
+	for _, r := range rules {
+		for _, p := range peers {
+			if e.peerRoutesAddr(p, r.TranslatedAddress) {
+				log.Infof("exclude forwarder peer from lazy connection: %s", p.GetWgPubKey())
+				excludedPeers[p.GetWgPubKey()] = true
+			}
 		}
 	}
+
 	return excludedPeers
 }

+// peerRoutesAddr reports whether the peer is a router for addr, matched against
+// the peer's already-parsed AllowedIPs from the store (the same typed value the
+// lazy manager consumes) rather than re-parsing the network map strings.
+func (e *Engine) peerRoutesAddr(p *mgmProto.RemotePeerConfig, addr netip.Addr) bool {
+	prefixes, ok := e.peerStore.AllowedIPs(p.GetWgPubKey())
+	if !ok {
+		return false
+	}
+	return prefixesContain(prefixes, addr)
+}
+
+// prefixesContain reports whether addr falls within any of the prefixes.
+func prefixesContain(prefixes []netip.Prefix, addr netip.Addr) bool {
+	for _, prefix := range prefixes {
+		if prefix.Contains(addr) {
+			return true
+		}
+	}
+	return false
+}
+
 // isChecksEqual checks if two slices of checks are equal.
 func isChecksEqual(checks1, checks2 []*mgmProto.Checks) bool {
 	normalize := func(checks []*mgmProto.Checks) []string {
--- a/client/internal/engine_lazy_exclude_test.go
+++ b/client/internal/engine_lazy_exclude_test.go
@@ -0,0 +1,87 @@
+package internal
+
+import (
+	"net/netip"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	firewallManager "github.com/netbirdio/netbird/client/firewall/manager"
+	"github.com/netbirdio/netbird/client/internal/peer"
+	"github.com/netbirdio/netbird/client/internal/peerstore"
+	mgmProto "github.com/netbirdio/netbird/shared/management/proto"
+)
+
+func TestPrefixesContain(t *testing.T) {
+	tests := []struct {
+		name     string
+		prefixes []string
+		addr     string
+		want     bool
+	}{
+		{name: "own overlay /32 matches", prefixes: []string{"100.110.8.145/32"}, addr: "100.110.8.145", want: true},
+		{name: "addr inside routed subnet", prefixes: []string{"10.121.0.0/16"}, addr: "10.121.208.4", want: true},
+		{name: "addr outside subnet", prefixes: []string{"10.121.0.0/16"}, addr: "10.122.0.1", want: false},
+		{name: "different /32", prefixes: []string{"100.110.8.145/32"}, addr: "100.110.8.146", want: false},
+		{name: "ipv6 /128 matches", prefixes: []string{"fd00::1/128"}, addr: "fd00::1", want: true},
+		{name: "no prefixes", prefixes: nil, addr: "10.121.208.4", want: false},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			prefixes := make([]netip.Prefix, 0, len(tt.prefixes))
+			for _, p := range tt.prefixes {
+				prefixes = append(prefixes, netip.MustParsePrefix(p))
+			}
+			require.Equal(t, tt.want, prefixesContain(prefixes, netip.MustParseAddr(tt.addr)))
+		})
+	}
+}
+
+// TestToExcludedLazyPeers_ForwardTarget guards a regression: the forward-target
+// peer (the peer routing a ForwardRule.TranslatedAddress) must be excluded from
+// lazy connections, matched via the peer's already-parsed AllowedIPs.
+func TestToExcludedLazyPeers_ForwardTarget(t *testing.T) {
+	const targetPeerKey = "cccccccccccccccccccccccccccccccccccccccccc0="
+	const otherPeerKey = "dddddddddddddddddddddddddddddddddddddddddd0="
+
+	store := peerstore.NewConnStore()
+	store.AddPeerConn(targetPeerKey, newTestConn(t, targetPeerKey, "100.110.8.145/32"))
+	store.AddPeerConn(otherPeerKey, newTestConn(t, otherPeerKey, "100.110.9.10/32"))
+
+	e := &Engine{peerStore: store}
+
+	peers := []*mgmProto.RemotePeerConfig{
+		{WgPubKey: targetPeerKey, AllowedIps: []string{"100.110.8.145/32"}},
+		{WgPubKey: otherPeerKey, AllowedIps: []string{"100.110.9.10/32"}},
+	}
+	rules := []firewallManager.ForwardRule{
+		{TranslatedAddress: netip.MustParseAddr("100.110.8.145")},
+	}
+
+	excluded := e.toExcludedLazyPeers(rules, peers)
+
+	require.True(t, excluded[targetPeerKey], "forward-target peer must be excluded from lazy connections")
+	require.False(t, excluded[otherPeerKey], "non-target peer must not be excluded")
+	require.Len(t, excluded, 1)
+}
+
+func TestToExcludedLazyPeers_NoRules(t *testing.T) {
+	e := &Engine{peerStore: peerstore.NewConnStore()}
+
+	peers := []*mgmProto.RemotePeerConfig{
+		{WgPubKey: "peer-a", AllowedIps: []string{"100.110.8.145/32"}},
+	}
+
+	require.Empty(t, e.toExcludedLazyPeers(nil, peers))
+}
+
+func newTestConn(t *testing.T, key, allowedIP string) *peer.Conn {
+	t.Helper()
+	conn, err := peer.NewConn(peer.ConnConfig{
+		Key:      key,
+		WgConfig: peer.WgConfig{AllowedIps: []netip.Prefix{netip.MustParsePrefix(allowedIP)}},
+	}, peer.ServiceDependencies{})
+	require.NoError(t, err)
+	return conn
+}
--- a/client/internal/engine_test.go
+++ b/client/internal/engine_test.go
@@ -279,8 +279,7 @@ func TestEngine_UpdateNetworkMap(t *testing.T) {
 	}, MobileDependency{})

 	wgIface := &MockWGIface{
-		NameFunc:            func() string { return "utun102" },
-		IsUserspaceBindFunc: func() bool { return true },
+		NameFunc: func() string { return "utun102" },
 		RemovePeerFunc: func(peerKey string) error {
 			return nil
 		},
--- a/client/server/server.go
+++ b/client/server/server.go
@@ -181,7 +181,7 @@ func (s *Server) Start() error {
 		log.Warnf("failed to redirect stderr: %v", err)
 	}

-	if err := restoreResidualState(s.rootCtx, s.profileManager.GetStatePath()); err != nil {
+	if err := RestoreResidualState(s.rootCtx, s.profileManager.GetStatePath()); err != nil {
 		log.Warnf(errRestoreResidualState, err)
 	}

@@ -551,7 +551,7 @@ func (s *Server) Login(callerCtx context.Context, msg *proto.LoginRequest) (*pro
 	s.actCancel = cancel
 	s.mutex.Unlock()

-	if err := restoreResidualState(s.rootCtx, s.profileManager.GetStatePath()); err != nil {
+	if err := RestoreResidualState(s.rootCtx, s.profileManager.GetStatePath()); err != nil {
 		log.Warnf(errRestoreResidualState, err)
 	}

@@ -858,7 +858,7 @@ func (s *Server) Up(callerCtx context.Context, msg *proto.UpRequest) (*proto.UpR

 		return s.waitForUp(callerCtx)
 	}
-	if err := restoreResidualState(callerCtx, s.profileManager.GetStatePath()); err != nil {
+	if err := RestoreResidualState(callerCtx, s.profileManager.GetStatePath()); err != nil {
 		log.Warnf(errRestoreResidualState, err)
 	}

--- a/client/server/state.go
+++ b/client/server/state.go
@@ -46,7 +46,7 @@ func (s *Server) CleanState(ctx context.Context, req *proto.CleanStateRequest) (

 	if req.All {
 		// Reuse existing cleanup logic for all states
-		if err := restoreResidualState(ctx, statePath); err != nil {
+		if err := RestoreResidualState(ctx, statePath); err != nil {
 			return nil, status.Errorf(codes.Internal, "failed to clean all states: %v", err)
 		}

@@ -113,9 +113,9 @@ func (s *Server) DeleteState(ctx context.Context, req *proto.DeleteStateRequest)
 	}, nil
 }

-// restoreResidualState checks if the client was not shut down in a clean way and restores residual if required.
+// RestoreResidualState checks if the client was not shut down in a clean way and restores residual if required.
 // Otherwise, we might not be able to connect to the management server to retrieve new config.
-func restoreResidualState(ctx context.Context, statePath string) error {
+func RestoreResidualState(ctx context.Context, statePath string) error {
 	if statePath == "" {
 		return nil
 	}
--- a/e2e/agentnetwork/chat_test.go
+++ b/e2e/agentnetwork/chat_test.go
@@ -91,7 +91,7 @@ func availableProviders() []providerCase {
 		if region == "" {
 			region = "us-east-1"
 		}
-		ps = append(ps, providerCase{name: "bedrock", catalogID: "bedrock_api", upstream: "https://bedrock-runtime." + region + ".amazonaws.com", apiKey: k, model: "us.anthropic.claude-haiku-4-5", kind: harness.WireMessages})
+		ps = append(ps, providerCase{name: "bedrock", catalogID: "bedrock_api", upstream: "https://bedrock-runtime." + region + ".amazonaws.com", apiKey: k, model: "us.anthropic.claude-haiku-4-5", kind: harness.WireBedrock})
 	}
 	return ps
 }
@@ -224,9 +224,12 @@ func TestProvidersMatrix(t *testing.T) {
 				var c int
 				var b string
 				var cerr error
-				if pc.kind == harness.WireVertex {
+				switch pc.kind {
+				case harness.WireVertex:
 					c, b, cerr = cl.Vertex(ctx, settings.Endpoint, proxyIP, pc.project, pc.region, pc.model, "Reply with exactly: pong", sessionID)
-				} else {
+				case harness.WireBedrock:
+					c, b, cerr = cl.Bedrock(ctx, settings.Endpoint, proxyIP, pc.model, "Reply with exactly: pong", sessionID)
+				default:
 					c, b, cerr = cl.Chat(ctx, settings.Endpoint, proxyIP, pc.kind, pc.model, "Reply with exactly: pong", sessionID)
 				}
 				if cerr == nil {
--- a/e2e/agentnetwork/guardrail_test.go
+++ b/e2e/agentnetwork/guardrail_test.go
@@ -0,0 +1,168 @@
+//go:build e2e
+
+package agentnetwork
+
+import (
+	"context"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/netbirdio/netbird/e2e/harness"
+	"github.com/netbirdio/netbird/shared/management/http/api"
+)
+
+// catalogModel returns the normalized catalog id the proxy stamps for a
+// path-routed provider's configured model — the form the guardrail allowlist is
+// compared against (region prefix / @version stripped).
+func catalogModel(pc providerCase) string {
+	switch pc.kind {
+	case harness.WireBedrock:
+		return strings.TrimPrefix(pc.model, "us.")
+	case harness.WireVertex:
+		return strings.SplitN(pc.model, "@", 2)[0]
+	default:
+		return pc.model
+	}
+}
+
+// disallowedModel returns a valid-shaped model id for the provider that is NOT
+// the configured/allowed one, so the guardrail must reject it before the
+// request ever reaches the upstream.
+func disallowedModel(pc providerCase) string {
+	switch pc.kind {
+	case harness.WireBedrock:
+		return "us.anthropic.claude-opus-4-8"
+	case harness.WireVertex:
+		return "claude-opus-4-8@20250101"
+	default:
+		return "unlisted-model"
+	}
+}
+
+// sendModel drives one request for the given model through the provider's native
+// wire shape and returns the HTTP status.
+func sendModel(ctx context.Context, t *testing.T, cl *harness.Client, endpoint, proxyIP string, pc providerCase, model string) int {
+	t.Helper()
+	var code int
+	var err error
+	switch pc.kind {
+	case harness.WireBedrock:
+		code, _, err = cl.Bedrock(ctx, endpoint, proxyIP, model, "Reply with exactly: pong", "")
+	case harness.WireVertex:
+		code, _, err = cl.Vertex(ctx, endpoint, proxyIP, pc.project, pc.region, model, "Reply with exactly: pong", "")
+	default:
+		code, _, err = cl.Chat(ctx, endpoint, proxyIP, pc.kind, model, "Reply with exactly: pong", "")
+	}
+	require.NoError(t, err, "request must reach the proxy for %s", pc.name)
+	return code
+}
+
+// TestModelAllowlistEnforced provisions a Model Allowlist guardrail limiting each
+// path-routed provider (Bedrock, Vertex) to its configured model, then drives
+// requests over the tunnel: the allowed model returns 200 while a model outside
+// the allowlist is denied 403 by the guardrail before it reaches the upstream.
+// This is the coverage missing for #6751 — the model for these providers travels
+// in the URL path, and the allowlist must be enforced there.
+func TestModelAllowlistEnforced(t *testing.T) {
+	var providers []providerCase
+	for _, pc := range availableProviders() {
+		if pc.kind == harness.WireBedrock || pc.kind == harness.WireVertex {
+			providers = append(providers, pc)
+		}
+	}
+	if len(providers) == 0 {
+		t.Skip("no path-routed provider keys set (AWS_BEARER_TOKEN_BEDROCK / GOOGLE_VERTEX_*); source ~/.llm-keys")
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Minute)
+	defer cancel()
+
+	grp, err := srv.API().Groups.Create(ctx, api.PostApiGroupsJSONRequestBody{Name: "e2e-allowlist"})
+	require.NoError(t, err, "create group")
+	t.Cleanup(func() { _ = srv.API().Groups.Delete(context.Background(), grp.Id) })
+
+	ephemeral := false
+	sk, err := srv.API().SetupKeys.Create(ctx, api.PostApiSetupKeysJSONRequestBody{
+		Name:       "e2e-allowlist-client",
+		Type:       "reusable",
+		ExpiresIn:  86400,
+		UsageLimit: 0,
+		AutoGroups: []string{grp.Id},
+		Ephemeral:  &ephemeral,
+	})
+	require.NoError(t, err, "mint setup key")
+
+	// Providers with their configured (allowed) models; the first bootstraps the cluster.
+	ids := make([]string, 0, len(providers))
+	allowed := make([]string, 0, len(providers))
+	for i, pc := range providers {
+		req := providerRequest(pc)
+		if i == 0 {
+			req.BootstrapCluster = ptr(harness.AgentNetworkCluster)
+		}
+		prov, perr := srv.CreateProvider(ctx, req)
+		require.NoError(t, perr, "create provider %s", pc.name)
+		id := prov.Id
+		ids = append(ids, id)
+		allowed = append(allowed, catalogModel(pc))
+		t.Cleanup(func() { _ = srv.DeleteProvider(context.Background(), id) })
+	}
+
+	// Guardrail allowlisting exactly the configured models.
+	var gr api.AgentNetworkGuardrailRequest
+	gr.Name = "e2e-allowlist"
+	gr.Checks.ModelAllowlist.Enabled = true
+	gr.Checks.ModelAllowlist.Models = allowed
+	guard, err := srv.CreateGuardrail(ctx, gr)
+	require.NoError(t, err, "create guardrail")
+	t.Cleanup(func() { _ = srv.DeleteGuardrail(context.Background(), guard.Id) })
+
+	enabled := true
+	pol, err := srv.CreatePolicy(ctx, api.AgentNetworkPolicyRequest{
+		Name:                   "e2e-allowlist",
+		Enabled:                &enabled,
+		SourceGroups:           []string{grp.Id},
+		DestinationProviderIds: ids,
+		GuardrailIds:           &[]string{guard.Id},
+	})
+	require.NoError(t, err, "create policy")
+	t.Cleanup(func() { _ = srv.DeletePolicy(context.Background(), pol.Id) })
+
+	settings, err := srv.GetSettings(ctx)
+	require.NoError(t, err, "read settings for endpoint")
+	require.NotEmpty(t, settings.Endpoint, "agent-network endpoint must be assigned")
+
+	proxyToken, err := srv.CreateProxyTokenCLI(ctx, "e2e-proxy-allowlist")
+	require.NoError(t, err, "mint proxy token via CLI")
+	px, err := harness.StartProxy(ctx, srv, proxyToken)
+	require.NoError(t, err, "start proxy")
+	t.Cleanup(func() { _ = px.Terminate(context.Background()) })
+
+	cl, err := harness.StartClient(ctx, srv, sk.Key)
+	require.NoError(t, err, "start client")
+	t.Cleanup(func() { _ = cl.Terminate(context.Background()) })
+
+	require.NoError(t, cl.WaitConnected(ctx, 90*time.Second), "client must connect to management")
+	if err := cl.WaitProxyPeer(ctx, 180*time.Second); err != nil {
+		t.Fatalf("client did not see the proxy peer: %v\n=== proxy logs ===\n%s", err, px.Logs(context.Background()))
+	}
+	proxyIP, err := cl.ResolveProxyIP(ctx, settings.Endpoint)
+	require.NoError(t, err, "resolve agent-network endpoint to proxy IP")
+
+	for _, pc := range providers {
+		pc := pc
+		t.Run(pc.name, func(t *testing.T) {
+			// The admin's allowlisted model is served end to end.
+			assert.Equal(t, 200, sendModel(ctx, t, cl, settings.Endpoint, proxyIP, pc, pc.model),
+				"allowlisted model must be permitted for %s", pc.name)
+			// A model outside the allowlist is rejected by the guardrail (before
+			// the upstream), regardless of whether it is a real catalog model.
+			assert.Equal(t, 403, sendModel(ctx, t, cl, settings.Endpoint, proxyIP, pc, disallowedModel(pc)),
+				"model outside the allowlist must be denied for %s", pc.name)
+		})
+	}
+}
--- a/e2e/harness/agentnetwork.go
+++ b/e2e/harness/agentnetwork.go
@@ -107,6 +107,17 @@ func (c *Combined) DeletePolicy(ctx context.Context, id string) error {
 	return anDelete(ctx, c, "/api/agent-network/policies/"+id)
 }

+// CreateGuardrail creates an agent-network guardrail (e.g. a model allowlist)
+// that can then be attached to a policy via its GuardrailIds.
+func (c *Combined) CreateGuardrail(ctx context.Context, req api.AgentNetworkGuardrailRequest) (api.AgentNetworkGuardrail, error) {
+	return anRequest[api.AgentNetworkGuardrail](ctx, c, http.MethodPost, "/api/agent-network/guardrails", req)
+}
+
+// DeleteGuardrail removes a guardrail by id.
+func (c *Combined) DeleteGuardrail(ctx context.Context, id string) error {
+	return anDelete(ctx, c, "/api/agent-network/guardrails/"+id)
+}
+
 // GetSettings returns the account's agent-network settings row. It exists only
 // after the first provider create bootstraps it.
 func (c *Combined) GetSettings(ctx context.Context) (api.AgentNetworkSettings, error) {
--- a/e2e/harness/client.go
+++ b/e2e/harness/client.go
@@ -194,6 +194,11 @@ const (
 	// WireVertex is the Anthropic-on-Vertex rawPredict shape: the client posts
 	// the full Vertex model path and the proxy mints the SA OAuth token.
 	WireVertex = "vertex"
+	// WireBedrock is the native AWS Bedrock InvokeModel shape: the model id
+	// travels in the URL path (/model/{id}/invoke), not the body, so the proxy
+	// routes by path. This is what a Bedrock SDK client sends and the shape the
+	// model-allowlist guardrail must enforce.
+	WireBedrock = "bedrock"
 )

 // Chat issues a chat-completion POST to the agent-network endpoint over the
@@ -226,6 +231,17 @@ func (cl *Client) Vertex(ctx context.Context, endpoint, proxyIP, project, region
 	return cl.post(ctx, endpoint, proxyIP, path, body, withSessionID(nil, sessionID))
 }

+// Bedrock issues a native AWS Bedrock InvokeModel POST over the tunnel. The
+// model id is carried in the request path (/model/{id}/invoke), so the proxy
+// routes by path; the body uses the bedrock anthropic_version rather than a
+// model field. A non-empty sessionID is sent as the universal x-session-id
+// header the proxy records.
+func (cl *Client) Bedrock(ctx context.Context, endpoint, proxyIP, model, prompt, sessionID string) (int, string, error) {
+	path := "/model/" + model + "/invoke"
+	body := fmt.Sprintf(`{"anthropic_version":"bedrock-2023-05-31","max_tokens":64,"messages":[{"role":"user","content":%q}]}`, prompt)
+	return cl.post(ctx, endpoint, proxyIP, path, body, withSessionID(nil, sessionID))
+}
+
 // withSessionID appends the x-session-id header when sessionID is non-empty.
 func withSessionID(headers []string, sessionID string) []string {
 	if sessionID == "" {
--- a/go.mod
+++ b/go.mod
@@ -103,6 +103,7 @@ require (
 	github.com/rs/xid v1.3.0
 	github.com/shirou/gopsutil/v4 v4.25.8
 	github.com/skratchdot/open-golang v0.0.0-20200116055534-eef842397966
+	github.com/soheilhy/cmux v0.1.5
 	github.com/songgao/water v0.0.0-20200317203138-2b4b6d7c09d8
 	github.com/stretchr/testify v1.11.1
 	github.com/testcontainers/testcontainers-go v0.37.0
--- a/go.sum
+++ b/go.sum
@@ -603,6 +603,8 @@ github.com/sirupsen/logrus v1.9.4 h1:TsZE7l11zFCLZnZ+teH4Umoq5BhEIfIzfRDZ1Uzql2w
 github.com/sirupsen/logrus v1.9.4/go.mod h1:ftWc9WdOfJ0a92nsE2jF5u5ZwH8Bv2zdeOC42RjbV2g=
 github.com/skratchdot/open-golang v0.0.0-20200116055534-eef842397966 h1:JIAuq3EEf9cgbU6AtGPK4CTG3Zf6CKMNqf0MHTggAUA=
 github.com/skratchdot/open-golang v0.0.0-20200116055534-eef842397966/go.mod h1:sUM3LWHvSMaG192sy56D9F7CNvL7jUJVXoqM1QKLnog=
+github.com/soheilhy/cmux v0.1.5 h1:jjzc5WVemNEDTLwv9tlmemhC73tI08BNOIGwBOo10Js=
+github.com/soheilhy/cmux v0.1.5/go.mod h1:T7TcVDs9LWfQgPlPsdngu6I6QIoyIFZDDC6sNE1GqG0=
 github.com/songgao/water v0.0.0-20200317203138-2b4b6d7c09d8 h1:TG/diQgUe0pntT/2D9tmUCz4VNwm9MfrtPr0SU2qSX8=
 github.com/songgao/water v0.0.0-20200317203138-2b4b6d7c09d8/go.mod h1:P5HUIBuIWKbyjl083/loAegFkfbFNx5i2qEP4CNbm7E=
 github.com/spf13/cast v1.10.0 h1:h2x0u2shc1QuLHfxi+cTJvs30+ZAHOGRic8uyGTDWxY=
@@ -757,6 +759,7 @@ golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLL
 golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20200520004742-59133d7f0dd7/go.mod h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A=
 golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
+golang.org/x/net v0.0.0-20201202161906-c7110b5ffcbb/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
 golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
 golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM=
 golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
--- a/management/internals/modules/peers/manager.go
+++ b/management/internals/modules/peers/manager.go
@@ -226,30 +226,6 @@ func (m *managerImpl) CreateProxyPeer(ctx context.Context, accountID string, pee
 		return nil
 	}

-	// Dedupe stale embedded peer records for the same (account, cluster).
-	// The proxy generates a fresh WireGuard keypair on every startup
-	// (proxy/internal/roundtrip/netbird.go), so without this sweep the
-	// prior embedded peer would linger forever — holding its CGNAT IP
-	// allocation, polluting other peers' rosters, and (most visibly)
-	// leaving the synth DNS pointing at the dead address. The
-	// (account, cluster) tuple identifies "the embedded peer for this
-	// proxy instance at this cluster"; any record matching that tuple
-	// with a different pubkey is by definition stale and must go.
-	staleIDs, err := m.findStaleEmbeddedProxyPeers(ctx, accountID, cluster, peerKey)
-	if err != nil {
-		return fmt.Errorf("scan for stale embedded proxy peers: %w", err)
-	}
-	if len(staleIDs) > 0 {
-		// userID="" + checkConnected=false: the deletion is initiated
-		// by management itself on behalf of the freshly-registering
-		// proxy, not by an end user; the stale peer may still be
-		// marked Connected from its prior session, but its session is
-		// dead by definition (its key no longer exists).
-		if err := m.DeletePeers(ctx, accountID, staleIDs, "", false); err != nil {
-			return fmt.Errorf("delete stale embedded proxy peers %v: %w", staleIDs, err)
-		}
-	}
-
 	name := fmt.Sprintf("proxy-%s", xid.New().String())
 	newPeer := &peer.Peer{
 		Ephemeral: true,
@@ -275,29 +251,3 @@ func (m *managerImpl) CreateProxyPeer(ctx context.Context, accountID string, pee

 	return nil
 }
-
-// findStaleEmbeddedProxyPeers returns the peer IDs of embedded proxy peer
-// records in accountID that target the same cluster but carry a different
-// WireGuard pubkey than the freshly-registering one. Used by CreateProxyPeer
-// to garbage-collect stale records left behind when the proxy restarts with a
-// regenerated keypair.
-func (m *managerImpl) findStaleEmbeddedProxyPeers(ctx context.Context, accountID, cluster, newKey string) ([]string, error) {
-	account, err := m.store.GetAccount(ctx, accountID)
-	if err != nil {
-		return nil, err
-	}
-	var stale []string
-	for _, p := range account.Peers {
-		if p == nil || !p.ProxyMeta.Embedded {
-			continue
-		}
-		if p.ProxyMeta.Cluster != cluster {
-			continue
-		}
-		if p.Key == newKey {
-			continue
-		}
-		stale = append(stale, p.ID)
-	}
-	return stale, nil
-}
--- a/management/internals/server/boot.go
+++ b/management/internals/server/boot.go
@@ -24,13 +24,13 @@ import (

 	"github.com/netbirdio/netbird/encryption"
 	"github.com/netbirdio/netbird/formatter/hook"
+	"github.com/netbirdio/netbird/management/internals/modules/agentnetwork"
 	"github.com/netbirdio/netbird/management/internals/modules/reverseproxy/accesslogs"
 	accesslogsmanager "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/accesslogs/manager"
 	rpservice "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/service"
 	nbgrpc "github.com/netbirdio/netbird/management/internals/shared/grpc"
 	"github.com/netbirdio/netbird/management/server/activity"
 	activitystore "github.com/netbirdio/netbird/management/server/activity/store"
-	"github.com/netbirdio/netbird/management/internals/modules/agentnetwork"
 	nbcache "github.com/netbirdio/netbird/management/server/cache"
 	nbContext "github.com/netbirdio/netbird/management/server/context"
 	nbhttp "github.com/netbirdio/netbird/management/server/http"
@@ -184,7 +184,12 @@ func (s *BaseServer) GRPCServer() *grpc.Server {
 			grpc.ChainStreamInterceptor(realip.StreamServerInterceptorOpts(realipOpts...), streamInterceptor, proxyStream),
 		}

-		if s.Config.HttpConfig.LetsEncryptDomain != "" {
+		// With the native transport enabled, TLS is terminated at the listeners
+		// (cmux-split shared listener and legacy port), so transport credentials
+		// must not be set or the server would attempt a second handshake.
+		if nativeGRPCEnabled() { //nolint:gocritic
+			log.Info("native gRPC transport enabled, TLS is terminated at the listeners")
+		} else if s.Config.HttpConfig.LetsEncryptDomain != "" {
 			certManager, err := encryption.CreateCertManager(s.Config.Datadir, s.Config.HttpConfig.LetsEncryptDomain)
 			if err != nil {
 				log.Fatalf("failed to create certificate service: %v", err)
--- a/management/internals/server/server.go
+++ b/management/internals/server/server.go
@@ -6,12 +6,16 @@ import (
 	"fmt"
 	"net"
 	"net/http"
+	"os"
+	"slices"
+	"strconv"
 	"strings"
 	"sync"
 	"time"

 	"github.com/google/uuid"
 	log "github.com/sirupsen/logrus"
+	"github.com/soheilhy/cmux"
 	"go.opentelemetry.io/otel/metric"
 	"golang.org/x/crypto/acme/autocert"
 	"golang.org/x/net/http2"
@@ -36,8 +40,18 @@ const (
 	DefaultSelfHostedDomain = "netbird.selfhosted"

 	ContainerKeyBaseServer = "baseServer"
+
+	// NativeGRPCEnvVar enables serving gRPC on the native gRPC transport,
+	// multiplexed with HTTP on the shared listener, instead of through the
+	// net/http ServeHTTP path which costs two extra goroutines per stream.
+	NativeGRPCEnvVar = "NB_MGMT_NATIVE_GRPC"
 )

+func nativeGRPCEnabled() bool {
+	enabled, _ := strconv.ParseBool(os.Getenv(NativeGRPCEnvVar))
+	return enabled
+}
+
 type Server interface {
 	Start(ctx context.Context) error
 	Stop() error
@@ -182,11 +196,22 @@ func (s *BaseServer) Start(ctx context.Context) error {
 		}
 	}

+	// With the native transport enabled the gRPC server carries no transport
+	// credentials, so TLS must be terminated at each of its listeners.
+	var grpcTLSConfig *tls.Config
+	if nativeGRPCEnabled() {
+		if s.certManager != nil {
+			grpcTLSConfig = s.certManager.TLSConfig()
+		} else {
+			grpcTLSConfig = tlsConfig
+		}
+	}
+
 	var compatListener net.Listener
 	if s.mgmtPort != ManagementLegacyPort && !s.disableLegacyManagementPort {
 		// The Management gRPC server was running on port 33073 previously. Old agents that are already connected to it
 		// are using port 33073. For compatibility purposes we keep running a 2nd gRPC server on port 33073.
-		compatListener, err = s.serveGRPC(srvCtx, s.GRPCServer(), ManagementLegacyPort)
+		compatListener, err = s.serveGRPC(srvCtx, s.GRPCServer(), ManagementLegacyPort, grpcTLSConfig)
 		if err != nil {
 			return err
 		}
@@ -196,22 +221,38 @@ func (s *BaseServer) Start(ctx context.Context) error {
 	rootHandler := s.handlerFunc(srvCtx, s.GRPCServer(), s.APIHandler(), s.IDPHandler(), s.Metrics().GetMeter())
 	switch {
 	case s.certManager != nil:
-		// a call to certManager.Listener() always creates a new listener so we do it once
-		cml := s.certManager.Listener()
 		if s.mgmtPort == 443 {
 			// CertManager, HTTP and gRPC API all on the same port
 			rootHandler = s.certManager.HTTPHandler(rootHandler)
-			s.listener = cml
+			if nativeGRPCEnabled() {
+				var tcpListener net.Listener
+				tcpListener, err = net.Listen("tcp", fmt.Sprintf(":%d", s.mgmtPort))
+				if err != nil {
+					return fmt.Errorf("failed creating TCP listener on port %d: %v", s.mgmtPort, err)
+				}
+				s.listener = tls.NewListener(tcpListener, preferHTTP1ForDualProtoClients(s.certManager.TLSConfig()))
+			} else {
+				s.listener = s.certManager.Listener()
+			}
 		} else {
-			s.listener, err = tls.Listen("tcp", fmt.Sprintf(":%d", s.mgmtPort), s.certManager.TLSConfig())
+			mgmtTLSConfig := s.certManager.TLSConfig()
+			if nativeGRPCEnabled() {
+				mgmtTLSConfig = preferHTTP1ForDualProtoClients(mgmtTLSConfig)
+			}
+			s.listener, err = tls.Listen("tcp", fmt.Sprintf(":%d", s.mgmtPort), mgmtTLSConfig)
 			if err != nil {
 				return fmt.Errorf("failed creating TLS listener on port %d: %v", s.mgmtPort, err)
 			}
+			cml := s.certManager.Listener()
 			log.WithContext(ctx).Infof("running HTTP server (LetsEncrypt challenge handler): %s", cml.Addr().String())
 			s.serveHTTP(ctx, cml, s.certManager.HTTPHandler(nil))
 		}
 	case tlsConfig != nil:
-		s.listener, err = tls.Listen("tcp", fmt.Sprintf(":%d", s.mgmtPort), tlsConfig)
+		mgmtTLSConfig := tlsConfig
+		if nativeGRPCEnabled() {
+			mgmtTLSConfig = preferHTTP1ForDualProtoClients(mgmtTLSConfig)
+		}
+		s.listener, err = tls.Listen("tcp", fmt.Sprintf(":%d", s.mgmtPort), mgmtTLSConfig)
 		if err != nil {
 			return fmt.Errorf("failed creating TLS listener on port %d: %v", s.mgmtPort, err)
 		}
@@ -224,7 +265,12 @@ func (s *BaseServer) Start(ctx context.Context) error {

 	log.WithContext(ctx).Infof("management server version %s", version.NetbirdVersion())
 	log.WithContext(ctx).Infof("running HTTP server and gRPC server on the same port: %s", s.listener.Addr().String())
-	s.serveGRPCWithHTTP(ctx, s.listener, rootHandler, tlsEnabled)
+	if nativeGRPCEnabled() {
+		log.WithContext(ctx).Infof("serving gRPC on the native transport (multiplexed with HTTP)")
+		s.serveMultiplexed(ctx, s.listener, s.GRPCServer(), rootHandler, tlsEnabled)
+	} else {
+		s.serveGRPCWithHTTP(ctx, s.listener, rootHandler, tlsEnabled)
+	}

 	s.update = version.NewUpdateAndStart("nb/management")
 	s.update.SetDaemonVersion(version.NetbirdVersion())
@@ -331,11 +377,14 @@ func (s *BaseServer) handlerFunc(_ context.Context, gRPCHandler *grpc.Server, ht
 	})
 }

-func (s *BaseServer) serveGRPC(ctx context.Context, grpcServer *grpc.Server, port int) (net.Listener, error) {
+func (s *BaseServer) serveGRPC(ctx context.Context, grpcServer *grpc.Server, port int, tlsConf *tls.Config) (net.Listener, error) {
 	listener, err := net.Listen("tcp", fmt.Sprintf(":%d", port))
 	if err != nil {
 		return nil, err
 	}
+	if tlsConf != nil {
+		listener = tls.NewListener(listener, tlsConf)
+	}

 	s.wg.Add(1)
 	go func() {
@@ -399,6 +448,69 @@ func (s *BaseServer) serveGRPCWithHTTP(ctx context.Context, listener net.Listene
 	}()
 }

+// preferHTTP1ForDualProtoClients steers TLS clients that offer both "h2" and
+// "http/1.1" in ALPN (browsers, REST clients) to HTTP/1.1. gRPC clients offer
+// only "h2", so with this steering every HTTP/2 connection on the shared
+// listener carries gRPC and can be routed to the native transport without
+// inspecting frames. ACME "acme-tls/1" and single-protocol clients keep the
+// base configuration.
+func preferHTTP1ForDualProtoClients(base *tls.Config) *tls.Config {
+	h1Config := base.Clone()
+	h1Config.NextProtos = []string{"http/1.1"}
+	steered := base.Clone()
+	steered.GetConfigForClient = func(hello *tls.ClientHelloInfo) (*tls.Config, error) {
+		if slices.Contains(hello.SupportedProtos, "http/1.1") && slices.Contains(hello.SupportedProtos, "h2") {
+			return h1Config, nil
+		}
+		return base, nil
+	}
+	return steered
+}
+
+// serveMultiplexed splits the shared listener by protocol: HTTP/2 connections
+// go to the native gRPC transport (see preferHTTP1ForDualProtoClients for why
+// they are all gRPC), everything else is served by net/http.
+//
+// Content-type based classification cannot be used here: cmux's SendSettings
+// matchers greet non-matching HTTP/2 connections and corrupt them for any
+// subsequent handler, while read-only matchers deadlock grpc-go clients,
+// which do not send HEADERS until they receive the server SETTINGS frame.
+func (s *BaseServer) serveMultiplexed(ctx context.Context, listener net.Listener, grpcServer *grpc.Server, handler http.Handler, tlsEnabled bool) {
+	mux := cmux.New(listener)
+	grpcListener := mux.Match(cmux.HTTP2())
+	httpListener := mux.Match(cmux.Any())
+
+	httpHandler := handler
+	if !tlsEnabled {
+		//nolint:staticcheck // h2c also handles the HTTP/1 Upgrade mechanism, which http.Server's UnencryptedHTTP2 does not
+		httpHandler = h2c.NewHandler(handler, &http2.Server{})
+	}
+
+	s.wg.Add(3)
+	go func() {
+		defer s.wg.Done()
+		s.reportServeError(ctx, grpcServer.Serve(grpcListener))
+	}()
+	go func() {
+		defer s.wg.Done()
+		s.reportServeError(ctx, http.Serve(httpListener, httpHandler))
+	}()
+	go func() {
+		defer s.wg.Done()
+		s.reportServeError(ctx, mux.Serve())
+	}()
+}
+
+func (s *BaseServer) reportServeError(ctx context.Context, err error) {
+	if ctx.Err() != nil || err == nil {
+		return
+	}
+	select {
+	case s.errCh <- err:
+	default:
+	}
+}
+
 // ResolveDomains determines dnsDomain and mgmtSingleAccModeDomain based on store state.
 // Fresh installs use the default self-hosted domain, while existing installs reuse the
 // persisted account domain to keep addressing stable across config changes.
--- a/management/internals/server/server_multiplex_test.go
+++ b/management/internals/server/server_multiplex_test.go
@@ -0,0 +1,109 @@
+package server
+
+import (
+	"context"
+	"crypto/ecdsa"
+	"crypto/elliptic"
+	"crypto/rand"
+	"crypto/tls"
+	"crypto/x509"
+	"crypto/x509/pkix"
+	"fmt"
+	"io"
+	"math/big"
+	"net"
+	"net/http"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/health"
+	healthpb "google.golang.org/grpc/health/grpc_health_v1"
+)
+
+func newSelfSignedCert(t *testing.T) tls.Certificate {
+	t.Helper()
+
+	key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
+	require.NoError(t, err)
+
+	template := x509.Certificate{
+		SerialNumber: big.NewInt(1),
+		Subject:      pkix.Name{CommonName: "127.0.0.1"},
+		NotBefore:    time.Now().Add(-time.Hour),
+		NotAfter:     time.Now().Add(time.Hour),
+		IPAddresses:  []net.IP{net.ParseIP("127.0.0.1")},
+	}
+	der, err := x509.CreateCertificate(rand.Reader, &template, &template, &key.PublicKey, key)
+	require.NoError(t, err)
+
+	return tls.Certificate{Certificate: [][]byte{der}, PrivateKey: key}
+}
+
+func TestServeMultiplexedRoutesProtocols(t *testing.T) {
+	tcpListener, err := net.Listen("tcp", "127.0.0.1:0")
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = tcpListener.Close() })
+
+	baseTLSConfig := &tls.Config{
+		Certificates: []tls.Certificate{newSelfSignedCert(t)},
+		NextProtos:   []string{"h2", "http/1.1"},
+	}
+	tlsListener := tls.NewListener(tcpListener, preferHTTP1ForDualProtoClients(baseTLSConfig))
+
+	grpcServer := grpc.NewServer()
+	healthpb.RegisterHealthServer(grpcServer, health.NewServer())
+	t.Cleanup(grpcServer.Stop)
+
+	handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		_, _ = fmt.Fprintf(w, "proto=%d", r.ProtoMajor)
+	})
+
+	s := &BaseServer{errCh: make(chan error, 4)}
+	ctx, cancel := context.WithCancel(context.Background())
+	t.Cleanup(cancel)
+	s.serveMultiplexed(ctx, tlsListener, grpcServer, handler, true)
+
+	addr := tcpListener.Addr().String()
+	url := "https://" + addr + "/"
+
+	grpcConn, err := grpc.NewClient(addr,
+		grpc.WithTransportCredentials(credentials.NewTLS(&tls.Config{InsecureSkipVerify: true})))
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = grpcConn.Close() })
+
+	checkCtx, checkCancel := context.WithTimeout(ctx, 5*time.Second)
+	defer checkCancel()
+	resp, err := healthpb.NewHealthClient(grpcConn).Check(checkCtx, &healthpb.HealthCheckRequest{})
+	require.NoError(t, err)
+	require.Equal(t, healthpb.HealthCheckResponse_SERVING, resp.Status)
+
+	get := func(client *http.Client) string {
+		t.Helper()
+		res, err := client.Get(url)
+		require.NoError(t, err)
+		body, err := io.ReadAll(res.Body)
+		require.NoError(t, err)
+		_ = res.Body.Close()
+		return string(body)
+	}
+
+	dualProtoClient := &http.Client{
+		Timeout: 5 * time.Second,
+		Transport: &http.Transport{
+			TLSClientConfig:   &tls.Config{InsecureSkipVerify: true},
+			ForceAttemptHTTP2: true,
+		},
+	}
+	require.Equal(t, "proto=1", get(dualProtoClient), "dual-ALPN client should be steered to HTTP/1.1")
+
+	h1OnlyClient := &http.Client{
+		Timeout: 5 * time.Second,
+		Transport: &http.Transport{
+			TLSClientConfig: &tls.Config{InsecureSkipVerify: true, NextProtos: []string{"http/1.1"}},
+		},
+	}
+	require.Equal(t, "proto=1", get(h1OnlyClient))
+}
--- a/management/internals/shared/grpc/conversion.go
+++ b/management/internals/shared/grpc/conversion.go
@@ -164,7 +164,6 @@ func ToSyncResponse(ctx context.Context, config *nbconfig.Config, httpConfig *nb
 	// filtered at the source (network map builder).
 	includeIPv6 := peer.SupportsIPv6() && peer.IPv6.IsValid()
 	useSourcePrefixes := peer.SupportsSourcePrefixes()
-	localIsProxy := peer.ProxyMeta.Embedded

 	response := &proto.SyncResponse{
 		PeerConfig: toPeerConfig(peer, networkMap.Network, dnsName, settings, httpConfig, deviceFlowConfig, networkMap.EnableSSH),
@@ -184,7 +183,7 @@ func ToSyncResponse(ctx context.Context, config *nbconfig.Config, httpConfig *nb
 	response.NetworkMap.PeerConfig = response.PeerConfig

 	remotePeers := make([]*proto.RemotePeerConfig, 0, len(networkMap.Peers)+len(networkMap.OfflinePeers))
-	remotePeers = appendRemotePeerConfig(remotePeers, networkMap.Peers, dnsName, includeIPv6, localIsProxy)
+	remotePeers = appendRemotePeerConfig(remotePeers, networkMap.Peers, dnsName, includeIPv6)

 	if !shouldSkipSendingDeprecatedRemotePeers(peer.Meta.WtVersion) {
 		response.RemotePeers = remotePeers
@@ -194,7 +193,7 @@ func ToSyncResponse(ctx context.Context, config *nbconfig.Config, httpConfig *nb
 	response.RemotePeersIsEmpty = len(remotePeers) == 0
 	response.NetworkMap.RemotePeersIsEmpty = response.RemotePeersIsEmpty

-	response.NetworkMap.OfflinePeers = appendRemotePeerConfig(nil, networkMap.OfflinePeers, dnsName, includeIPv6, localIsProxy)
+	response.NetworkMap.OfflinePeers = appendRemotePeerConfig(nil, networkMap.OfflinePeers, dnsName, includeIPv6)

 	firewallRules := toProtocolFirewallRules(networkMap.FirewallRules, includeIPv6, useSourcePrefixes)
 	response.NetworkMap.FirewallRules = firewallRules
@@ -293,7 +292,7 @@ func shouldSkipSendingDeprecatedRemotePeers(peerVersion string) bool {
 	return precomputedDeprecatedRemotePeersConstraint.Check(peerNBVersion)
 }

-func appendRemotePeerConfig(dst []*proto.RemotePeerConfig, peers []*nbpeer.Peer, dnsName string, includeIPv6 bool, localIsProxy bool) []*proto.RemotePeerConfig {
+func appendRemotePeerConfig(dst []*proto.RemotePeerConfig, peers []*nbpeer.Peer, dnsName string, includeIPv6 bool) []*proto.RemotePeerConfig {
 	for _, rPeer := range peers {
 		allowedIPs := []string{rPeer.IP.String() + "/32"}
 		if includeIPv6 && rPeer.IPv6.IsValid() {
@@ -305,24 +304,11 @@ func appendRemotePeerConfig(dst []*proto.RemotePeerConfig, peers []*nbpeer.Peer,
 			SshConfig:    &proto.SSHConfig{SshPubKey: []byte(rPeer.SSHKey)},
 			Fqdn:         rPeer.FQDN(dnsName),
 			AgentVersion: rPeer.Meta.WtVersion,
-			LazyState:    lazyStateFor(localIsProxy, rPeer),
 		})
 	}
 	return dst
 }

-// lazyStateFor returns the per-peer lazy override for a remote peer. Connections
-// involving an ephemeral proxy peer on either endpoint default to lazy so shared
-// proxy infrastructure is not kept permanently connected to every peer. All
-// other peers follow the account-wide flag. A future admin-facing per-peer
-// setting can return LazyStateEager here to force a peer always-active.
-func lazyStateFor(localIsProxy bool, rPeer *nbpeer.Peer) proto.LazyState {
-	if localIsProxy || rPeer.ProxyMeta.Embedded {
-		return proto.LazyState_LazyStateLazy
-	}
-	return proto.LazyState_LazyStateDefault
-}
-
 // toProtocolDNSConfig converts nbdns.Config to proto.DNSConfig using the cache
 func toProtocolDNSConfig(update nbdns.Config, cache *cache.DNSConfigCache, forwardPort int64) *proto.DNSConfig {
 	protoUpdate := &proto.DNSConfig{
--- a/management/internals/shared/grpc/server.go
+++ b/management/internals/shared/grpc/server.go
@@ -215,12 +215,13 @@ func (s *Server) Job(srv proto.ManagementService_JobServer) error {
 		return status.Errorf(codes.Unauthenticated, "peer is not registered")
 	}

-	s.startResponseReceiver(ctx, srv)
-
-	updates := s.jobManager.CreateJobChannel(ctx, accountID, peer.ID)
+	stream := s.jobManager.RegisterStream(ctx, accountID, peer.ID, func(event *job.Event) error {
+		return s.sendJob(ctx, peerKey, event, srv)
+	})
+	defer s.jobManager.UnregisterStream(ctx, accountID, peer.ID, stream)
 	log.WithContext(ctx).Debugf("Job: took %v", time.Since(reqStart))

-	return s.sendJobsLoop(ctx, accountID, peerKey, peer, updates, srv)
+	return s.receiveJobResponses(ctx, peerKey, srv)
 }

 // Sync validates the existence of a connecting peer, sends an initial state (all available for the connecting peers) and
@@ -362,51 +363,26 @@ func (s *Server) handleHandshake(ctx context.Context, srv proto.ManagementServic
 	return peerKey, nil
 }

-func (s *Server) startResponseReceiver(ctx context.Context, srv proto.ManagementService_JobServer) {
-	go func() {
-		for {
-			msg, err := srv.Recv()
-			if err != nil {
-				if errors.Is(err, io.EOF) || errors.Is(err, context.Canceled) {
-					return
-				}
-				log.WithContext(ctx).Warnf("recv job response error: %v", err)
-				return
-			}
-
-			jobResp := &proto.JobResponse{}
-			if _, err := s.parseRequest(ctx, msg, jobResp); err != nil {
-				log.WithContext(ctx).Warnf("invalid job response: %v", err)
-				continue
-			}
-
-			if err := s.jobManager.HandleResponse(ctx, jobResp, msg.WgPubKey); err != nil {
-				log.WithContext(ctx).Errorf("handle job response failed: %v", err)
-			}
-		}
-	}()
-}
-
-func (s *Server) sendJobsLoop(ctx context.Context, accountID string, peerKey wgtypes.Key, peer *nbpeer.Peer, updates *job.Channel, srv proto.ManagementService_JobServer) error {
-	// todo figure out better error handling strategy
-	defer s.jobManager.CloseChannel(ctx, accountID, peer.ID)
-
+func (s *Server) receiveJobResponses(ctx context.Context, peerKey wgtypes.Key, srv proto.ManagementService_JobServer) error {
 	for {
-		event, err := updates.Event(ctx)
+		msg, err := srv.Recv()
 		if err != nil {
-			if errors.Is(err, job.ErrJobChannelClosed) {
-				log.WithContext(ctx).Debugf("jobs channel for peer %s was closed", peerKey.String())
-				return nil
+			if errors.Is(err, io.EOF) || errors.Is(err, context.Canceled) || ctx.Err() != nil {
+				log.WithContext(ctx).Debugf("job stream of peer %s has been closed", peerKey.String())
+				return nil //nolint:nilerr
 			}
-
-			// happens when connection drops, e.g. client disconnects
-			log.WithContext(ctx).Debugf("stream of peer %s has been closed", peerKey.String())
-			return ctx.Err()
+			log.WithContext(ctx).Warnf("recv job response error: %v", err)
+			return err
 		}

-		if err := s.sendJob(ctx, peerKey, event, srv); err != nil {
-			log.WithContext(ctx).Warnf("send job failed: %v", err)
-			return nil
+		jobResp := &proto.JobResponse{}
+		if _, err := s.parseRequest(ctx, msg, jobResp); err != nil {
+			log.WithContext(ctx).Warnf("invalid job response: %v", err)
+			continue
+		}
+
+		if err := s.jobManager.HandleResponse(ctx, jobResp, msg.WgPubKey); err != nil {
+			log.WithContext(ctx).Errorf("handle job response failed: %v", err)
 		}
 	}
 }
--- a/management/internals/shared/grpc/token_mgr.go
+++ b/management/internals/shared/grpc/token_mgr.go
@@ -43,8 +43,9 @@ type TimeBasedAuthSecretsManager struct {
 	updateManager   network_map.PeersUpdateManager
 	settingsManager settings.Manager
 	groupsManager   groups.Manager
-	turnCancelMap   map[string]chan struct{}
-	relayCancelMap  map[string]chan struct{}
+	scheduler       *refreshScheduler
+	turnJobs        map[string]*refreshJob
+	relayJobs       map[string]*refreshJob
 	wgKey           wgtypes.Key
 }

@@ -60,8 +61,6 @@ func NewTimeBasedAuthSecretsManager(updateManager network_map.PeersUpdateManager
 		updateManager:   updateManager,
 		turnCfg:         turnCfg,
 		relayCfg:        relayCfg,
-		turnCancelMap:   make(map[string]chan struct{}),
-		relayCancelMap:  make(map[string]chan struct{}),
 		settingsManager: settingsManager,
 		groupsManager:   groupsManager,
 		wgKey:           key,
@@ -127,16 +126,16 @@ func (m *TimeBasedAuthSecretsManager) GenerateRelayToken() (*Token, error) {
 }

 func (m *TimeBasedAuthSecretsManager) cancelTURN(peerID string) {
-	if channel, ok := m.turnCancelMap[peerID]; ok {
-		close(channel)
-		delete(m.turnCancelMap, peerID)
+	if job, ok := m.turnJobs[peerID]; ok {
+		m.scheduler.cancel(job)
+		delete(m.turnJobs, peerID)
 	}
 }

 func (m *TimeBasedAuthSecretsManager) cancelRelay(peerID string) {
-	if channel, ok := m.relayCancelMap[peerID]; ok {
-		close(channel)
-		delete(m.relayCancelMap, peerID)
+	if job, ok := m.relayJobs[peerID]; ok {
+		m.scheduler.cancel(job)
+		delete(m.relayJobs, peerID)
 	}
 }

@@ -148,6 +147,31 @@ func (m *TimeBasedAuthSecretsManager) CancelRefresh(peerID string) {
 	m.cancelRelay(peerID)
 }

+func (m *TimeBasedAuthSecretsManager) ensureScheduler() {
+	if m.scheduler == nil {
+		m.scheduler = newRefreshScheduler(m.runRefreshJob)
+		m.turnJobs = make(map[string]*refreshJob)
+		m.relayJobs = make(map[string]*refreshJob)
+	}
+}
+
+func (m *TimeBasedAuthSecretsManager) runRefreshJob(job *refreshJob) {
+	switch job.kind {
+	case refreshKindTURN:
+		m.pushNewTURNAndRelayTokens(job.ctx, job.accountID, job.peerID)
+	case refreshKindRelay:
+		m.pushNewRelayTokens(job.ctx, job.accountID, job.peerID)
+	}
+}
+
+func refreshInterval(ttl time.Duration) time.Duration {
+	interval := ttl / 4 * 3
+	if interval <= 0 {
+		interval = defaultDuration / 4 * 3
+	}
+	return interval
+}
+
 // SetupRefresh starts peer credentials refresh
 func (m *TimeBasedAuthSecretsManager) SetupRefresh(ctx context.Context, accountID, peerID string) {
 	m.mux.Lock()
@@ -157,54 +181,38 @@ func (m *TimeBasedAuthSecretsManager) SetupRefresh(ctx context.Context, accountI
 	m.cancelRelay(peerID)

 	if m.turnCfg != nil && m.turnCfg.TimeBasedCredentials {
-		turnCancel := make(chan struct{}, 1)
-		m.turnCancelMap[peerID] = turnCancel
-		go m.refreshTURNTokens(ctx, accountID, peerID, turnCancel)
+		m.ensureScheduler()
+		job := &refreshJob{
+			ctx:       ctx,
+			accountID: accountID,
+			peerID:    peerID,
+			kind:      refreshKindTURN,
+			interval:  refreshInterval(m.turnCfg.CredentialsTTL.Duration),
+		}
+		m.turnJobs[peerID] = job
+		m.scheduler.schedule(job)
 		log.WithContext(ctx).Debugf("starting TURN refresh for %s", peerID)
 	} else {
 		log.WithContext(ctx).Debugf("no TURN configuration, skipping TURN refresh for %s", peerID)
 	}

 	if m.relayCfg != nil {
-		relayCancel := make(chan struct{}, 1)
-		m.relayCancelMap[peerID] = relayCancel
-		go m.refreshRelayTokens(ctx, accountID, peerID, relayCancel)
+		m.ensureScheduler()
+		job := &refreshJob{
+			ctx:       ctx,
+			accountID: accountID,
+			peerID:    peerID,
+			kind:      refreshKindRelay,
+			interval:  refreshInterval(m.relayCfg.CredentialsTTL.Duration),
+		}
+		m.relayJobs[peerID] = job
+		m.scheduler.schedule(job)
 		log.WithContext(ctx).Tracef("starting relay refresh for %s", peerID)
 	} else {
 		log.WithContext(ctx).Tracef("no relay configuration, skipping relay refresh for %s", peerID)
 	}
 }

-func (m *TimeBasedAuthSecretsManager) refreshTURNTokens(ctx context.Context, accountID, peerID string, cancel chan struct{}) {
-	ticker := time.NewTicker(m.turnCfg.CredentialsTTL.Duration / 4 * 3)
-	defer ticker.Stop()
-
-	for {
-		select {
-		case <-cancel:
-			log.WithContext(ctx).Tracef("stopping TURN refresh for %s", peerID)
-			return
-		case <-ticker.C:
-			m.pushNewTURNAndRelayTokens(ctx, accountID, peerID)
-		}
-	}
-}
-
-func (m *TimeBasedAuthSecretsManager) refreshRelayTokens(ctx context.Context, accountID, peerID string, cancel chan struct{}) {
-	ticker := time.NewTicker(m.relayCfg.CredentialsTTL.Duration / 4 * 3)
-	defer ticker.Stop()
-
-	for {
-		select {
-		case <-cancel:
-			log.WithContext(ctx).Tracef("stopping relay refresh for %s", peerID)
-			return
-		case <-ticker.C:
-			m.pushNewRelayTokens(ctx, accountID, peerID)
-		}
-	}
-}
-
 func (m *TimeBasedAuthSecretsManager) pushNewTURNAndRelayTokens(ctx context.Context, accountID, peerID string) {
 	turnToken, err := m.turnHmacToken.GenerateToken(sha1.New)
 	if err != nil {
--- a/management/internals/shared/grpc/token_mgr_test.go
+++ b/management/internals/shared/grpc/token_mgr_test.go
@@ -112,12 +112,12 @@ func TestTimeBasedAuthSecretsManager_SetupRefresh(t *testing.T) {

 	tested.SetupRefresh(ctx, "someAccountID", peer)

-	if _, ok := tested.turnCancelMap[peer]; !ok {
-		t.Errorf("expecting peer to be present in the turn cancel map, got not present")
+	if _, ok := tested.turnJobs[peer]; !ok {
+		t.Errorf("expecting peer to be present in the turn jobs map, got not present")
 	}

-	if _, ok := tested.relayCancelMap[peer]; !ok {
-		t.Errorf("expecting peer to be present in the relay cancel map, got not present")
+	if _, ok := tested.relayJobs[peer]; !ok {
+		t.Errorf("expecting peer to be present in the relay jobs map, got not present")
 	}

 	var updates []*network_map.UpdateMessage
@@ -212,19 +212,26 @@ func TestTimeBasedAuthSecretsManager_CancelRefresh(t *testing.T) {
 	require.NoError(t, err)

 	tested.SetupRefresh(context.Background(), "someAccountID", peer)
-	if _, ok := tested.turnCancelMap[peer]; !ok {
-		t.Errorf("expecting peer to be present in turn cancel map, got not present")
+	if _, ok := tested.turnJobs[peer]; !ok {
+		t.Errorf("expecting peer to be present in turn jobs map, got not present")
 	}
-	if _, ok := tested.relayCancelMap[peer]; !ok {
-		t.Errorf("expecting peer to be present in relay cancel map, got not present")
+	if _, ok := tested.relayJobs[peer]; !ok {
+		t.Errorf("expecting peer to be present in relay jobs map, got not present")
 	}

 	tested.CancelRefresh(peer)
-	if _, ok := tested.turnCancelMap[peer]; ok {
-		t.Errorf("expecting peer to be not present in turn cancel map, got present")
+	if _, ok := tested.turnJobs[peer]; ok {
+		t.Errorf("expecting peer to be not present in turn jobs map, got present")
 	}
-	if _, ok := tested.relayCancelMap[peer]; ok {
-		t.Errorf("expecting peer to be not present in relay cancel map, got present")
+	if _, ok := tested.relayJobs[peer]; ok {
+		t.Errorf("expecting peer to be not present in relay jobs map, got present")
+	}
+
+	tested.scheduler.mu.Lock()
+	heapLen := len(tested.scheduler.jobs)
+	tested.scheduler.mu.Unlock()
+	if heapLen != 0 {
+		t.Errorf("expecting scheduler heap to be empty after cancel, got %d entries", heapLen)
 	}
 }

--- a/management/internals/shared/grpc/token_refresh_scheduler.go
+++ b/management/internals/shared/grpc/token_refresh_scheduler.go
@@ -0,0 +1,158 @@
+package grpc
+
+import (
+	"container/heap"
+	"context"
+	"sync"
+	"sync/atomic"
+	"time"
+)
+
+const (
+	refreshWorkerCount   = 4
+	refreshWorkQueueSize = 1024
+)
+
+type refreshKind int
+
+const (
+	refreshKindTURN refreshKind = iota
+	refreshKindRelay
+)
+
+type refreshJob struct {
+	ctx       context.Context
+	accountID string
+	peerID    string
+	kind      refreshKind
+	interval  time.Duration
+	nextRun   time.Time
+	index     int
+	cancelled atomic.Bool
+}
+
+type refreshJobHeap []*refreshJob
+
+func (h refreshJobHeap) Len() int           { return len(h) }
+func (h refreshJobHeap) Less(i, j int) bool { return h[i].nextRun.Before(h[j].nextRun) }
+
+func (h refreshJobHeap) Swap(i, j int) {
+	h[i], h[j] = h[j], h[i]
+	h[i].index = i
+	h[j].index = j
+}
+
+func (h *refreshJobHeap) Push(x any) {
+	job := x.(*refreshJob)
+	job.index = len(*h)
+	*h = append(*h, job)
+}
+
+func (h *refreshJobHeap) Pop() any {
+	old := *h
+	n := len(old)
+	job := old[n-1]
+	old[n-1] = nil
+	job.index = -1
+	*h = old[:n-1]
+	return job
+}
+
+// refreshScheduler executes periodic credential refresh jobs for all peers
+// from one timer goroutine and a fixed worker pool, instead of two parked
+// goroutines per connected peer.
+type refreshScheduler struct {
+	mu   sync.Mutex
+	jobs refreshJobHeap
+	wake chan struct{}
+	work chan *refreshJob
+	run  func(job *refreshJob)
+}
+
+func newRefreshScheduler(run func(job *refreshJob)) *refreshScheduler {
+	s := &refreshScheduler{
+		wake: make(chan struct{}, 1),
+		work: make(chan *refreshJob, refreshWorkQueueSize),
+		run:  run,
+	}
+	go s.loop()
+	for range refreshWorkerCount {
+		go s.worker()
+	}
+	return s
+}
+
+func (s *refreshScheduler) schedule(job *refreshJob) {
+	s.mu.Lock()
+	job.nextRun = time.Now().Add(job.interval)
+	heap.Push(&s.jobs, job)
+	s.mu.Unlock()
+
+	select {
+	case s.wake <- struct{}{}:
+	default:
+	}
+}
+
+func (s *refreshScheduler) cancel(job *refreshJob) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	job.cancelled.Store(true)
+	if job.index >= 0 {
+		heap.Remove(&s.jobs, job.index)
+	}
+}
+
+func (s *refreshScheduler) loop() {
+	timer := time.NewTimer(time.Hour)
+	if !timer.Stop() {
+		<-timer.C
+	}
+
+	for {
+		s.mu.Lock()
+		now := time.Now()
+		var due []*refreshJob
+		for len(s.jobs) > 0 && !s.jobs[0].nextRun.After(now) {
+			job := s.jobs[0]
+			job.nextRun = job.nextRun.Add(job.interval)
+			if !job.nextRun.After(now) {
+				job.nextRun = now.Add(job.interval)
+			}
+			heap.Fix(&s.jobs, 0)
+			due = append(due, job)
+		}
+		wait := time.Duration(-1)
+		if len(s.jobs) > 0 {
+			wait = time.Until(s.jobs[0].nextRun)
+		}
+		s.mu.Unlock()
+
+		for _, job := range due {
+			s.work <- job
+		}
+
+		if wait < 0 {
+			<-s.wake
+			continue
+		}
+
+		timer.Reset(wait)
+		select {
+		case <-s.wake:
+			if !timer.Stop() {
+				<-timer.C
+			}
+		case <-timer.C:
+		}
+	}
+}
+
+func (s *refreshScheduler) worker() {
+	for job := range s.work {
+		if job.cancelled.Load() {
+			continue
+		}
+		s.run(job)
+	}
+}
--- a/management/internals/shared/grpc/token_refresh_scheduler_test.go
+++ b/management/internals/shared/grpc/token_refresh_scheduler_test.go
@@ -0,0 +1,56 @@
+package grpc
+
+import (
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestRefreshInterval(t *testing.T) {
+	defaultInterval := defaultDuration / 4 * 3
+
+	require.Equal(t, 9*time.Hour, refreshInterval(12*time.Hour))
+	require.Equal(t, defaultInterval, refreshInterval(0))
+	require.Equal(t, defaultInterval, refreshInterval(-time.Second))
+	require.Equal(t, defaultInterval, refreshInterval(3*time.Nanosecond))
+	require.Positive(t, refreshInterval(4*time.Nanosecond))
+}
+
+func TestWorkerSkipsCancelledJob(t *testing.T) {
+	var ran atomic.Int32
+	scheduler := newRefreshScheduler(func(*refreshJob) {
+		ran.Add(1)
+	})
+
+	cancelledJob := &refreshJob{interval: time.Hour}
+	cancelledJob.cancelled.Store(true)
+	liveJob := &refreshJob{interval: time.Hour}
+
+	scheduler.work <- cancelledJob
+	scheduler.work <- liveJob
+
+	require.Eventually(t, func() bool {
+		return ran.Load() == 1
+	}, 2*time.Second, 10*time.Millisecond, "live job should run exactly once, cancelled job never")
+}
+
+func TestCancelBeforeFirePreventsRun(t *testing.T) {
+	var ran atomic.Int32
+	scheduler := newRefreshScheduler(func(*refreshJob) {
+		ran.Add(1)
+	})
+
+	job := &refreshJob{interval: 50 * time.Millisecond}
+	scheduler.schedule(job)
+	scheduler.cancel(job)
+
+	time.Sleep(150 * time.Millisecond)
+	require.Zero(t, ran.Load(), "cancelled job must never fire")
+
+	scheduler.mu.Lock()
+	heapLen := len(scheduler.jobs)
+	scheduler.mu.Unlock()
+	require.Zero(t, heapLen)
+}
--- a/management/main.go
+++ b/management/main.go
@@ -1,19 +1,24 @@
 package main

 import (
-	"log"
 	"net/http"
 	// nolint:gosec
 	_ "net/http/pprof"
 	"os"

+	log "github.com/sirupsen/logrus"
+
 	"github.com/netbirdio/netbird/management/cmd"
 )

 func main() {
-	go func() {
-		log.Println(http.ListenAndServe("localhost:6060", nil))
-	}()
+	if pprofAddr := os.Getenv("NB_PPROF_ADDR"); pprofAddr != "" {
+		log.Infof("pprof enabled, listening on: %s", pprofAddr)
+		go func() {
+			log.Println(http.ListenAndServe(pprofAddr, nil))
+		}()
+	}
+
 	if err := cmd.Execute(); err != nil {
 		os.Exit(1)
 	}
--- a/management/server/agentnetwork_proxypeer_restart_test.go
+++ b/management/server/agentnetwork_proxypeer_restart_test.go
@@ -1,199 +0,0 @@
-package server
-
-import (
-	"context"
-	"testing"
-	"time"
-
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-
-	"github.com/netbirdio/netbird/management/internals/modules/peers"
-	"github.com/netbirdio/netbird/management/internals/modules/agentnetwork"
-	agenttypes "github.com/netbirdio/netbird/management/internals/modules/agentnetwork/types"
-	nbpeer "github.com/netbirdio/netbird/management/server/peer"
-	"github.com/netbirdio/netbird/management/server/permissions"
-	"github.com/netbirdio/netbird/management/server/store"
-	"github.com/netbirdio/netbird/management/server/types"
-)
-
-// TestAgentNetwork_ProxyRestart_PropagatesNewPeerAndDropsStale is the no-mock
-// regression guard for the bug the user reported: restarting the proxy creates
-// a fresh embedded peer with a NEW WireGuard public key (the proxy generates
-// the keypair on every startup at proxy/internal/roundtrip/netbird.go:312).
-// The PRIOR embedded peer record is never deleted on management, so the
-// account accumulates a stale peer holding a stale CGNAT IP. Other peers
-// in the account either keep routing to the dead IP, or — if synth DNS
-// picks the wrong record — never see the new IP at all.
-//
-// What this test exercises (no mocks):
-//   - real SQLite test store
-//   - real DefaultAccountManager, network-map controller, peer-update channels
-//   - real peers.Manager.CreateProxyPeer path (the very method the proxy
-//     invokes over gRPC on every startup)
-//   - real agentnetwork.Manager + synth chain so the client receives a
-//     concrete DNS record that must point at the LATEST proxy peer.
-//
-// Pre-fix expected behavior (red): two embedded peers exist after the
-// "restart"; the synth DNS record points at the stale one; the client
-// receives an update reflecting the new peer but the old one lingers.
-// Post-fix expected behavior (green): exactly one embedded peer exists
-// after restart (with the new key) AND the client's network map carries
-// the synth DNS pointing at that new peer's CGNAT IP.
-func TestAgentNetwork_ProxyRestart_PropagatesNewPeerAndDropsStale(t *testing.T) {
-	am, updateManager, err := createManager(t)
-	require.NoError(t, err, "createManager must succeed")
-	ctx := context.Background()
-
-	const (
-		accountID   = "an-restart-acct"
-		adminUserID = "an-restart-admin"
-		groupAID    = "an-restart-grp-A"
-		clusterAddr = "eu.proxy.netbird.io"
-		clientKey   = "BhRPtynAAYRDy08+q4HTMsos8fs4plTP4NOSh7C1ry8="
-		// Two different proxy pubkeys — the "before" and "after" of a
-		// proxy-process restart with fresh-keypair generation.
-		proxyKey1 = "Aaaaa1aaaaYRDy08+q4HTMsos8fs4plTP4NOSh7C1ry8="
-		proxyKey2 = "Bbbbb2bbbbYRDy08+q4HTMsos8fs4plTP4NOSh7C1ry8="
-	)
-
-	// --- Account scaffold ---
-	account := newAccountWithId(ctx, accountID, adminUserID, "an-restart.test", "", "", false)
-	require.NoError(t, am.Store.SaveAccount(ctx, account))
-
-	clientPeer := &nbpeer.Peer{
-		Key:      clientKey,
-		Name:     "an-restart-client",
-		DNSLabel: "an-restart-client",
-		Meta:     nbpeer.PeerSystemMeta{Hostname: "an-restart-client", GoOS: "linux", WtVersion: "development"},
-	}
-	addedClient, _, _, _, err := am.AddPeer(ctx, "", "", adminUserID, clientPeer, false)
-	require.NoError(t, err, "AddPeer for client must succeed")
-	require.NoError(t, am.MarkPeerConnected(ctx, clientKey, accountID, time.Now().UnixNano(), &types.NetworkMap{}),
-		"MarkPeerConnected for the client peer must succeed (affected-peer fan-out skips disconnected peers)")
-
-	// Place the client in group A so the synth policy reaches it.
-	account, err = am.Store.GetAccount(ctx, accountID)
-	require.NoError(t, err)
-	account.Groups[groupAID] = &types.Group{ID: groupAID, Name: "groupA", Peers: []string{addedClient.ID}}
-	require.NoError(t, am.Store.SaveAccount(ctx, account), "SaveAccount must persist group A")
-
-	// --- Real peers + agent-network managers ---
-	permMgr := permissions.NewManager(am.Store)
-	peersMgr := peers.NewManager(am.Store, permMgr)
-	peersMgr.SetAccountManager(am)
-	peersMgr.SetNetworkMapController(am.networkMapController)
-	agentMgr := agentnetwork.NewManager(am.Store, permMgr, am, nil)
-
-	// Subscribe BEFORE any state-mutating call so we don't lose the update
-	// that contains the synth DNS record.
-	clientCh := updateManager.CreateChannel(ctx, addedClient.ID)
-	t.Cleanup(func() { updateManager.CloseChannel(ctx, addedClient.ID) })
-	drain(clientCh)
-
-	// --- First proxy startup: register peer key K1, then mark it
-	// connected. In production the proxy follows CreateProxyPeer with the
-	// regular sync stream which lands on MarkPeerConnected; the synth DNS
-	// path filters out peers that aren't Connected (types/account.go:323),
-	// so without this step no DNS record would be emitted.
-	require.NoError(t, peersMgr.CreateProxyPeer(ctx, accountID, proxyKey1, clusterAddr),
-		"first CreateProxyPeer (proxy startup) must succeed")
-
-	peer1ID, err := am.Store.GetPeerIDByKey(ctx, store.LockingStrengthNone, proxyKey1)
-	require.NoError(t, err, "proxy peer for K1 must be persisted after CreateProxyPeer")
-	require.NotEmpty(t, peer1ID)
-
-	require.NoError(t, am.MarkPeerConnected(ctx, proxyKey1, accountID, time.Now().UnixNano(), &types.NetworkMap{}),
-		"MarkPeerConnected for K1 must succeed")
-
-	account, err = am.Store.GetAccount(ctx, accountID)
-	require.NoError(t, err)
-	proxyIP1 := account.Peers[peer1ID].IP.String()
-	require.NotEmpty(t, proxyIP1, "K1 must have an assigned overlay IP")
-
-	// --- Provider + policy. CreateProvider / CreatePolicy trigger the
-	// agentnetwork reconcile which runs UpdateAccountPeers; the resulting
-	// NetworkMap delivered to the client carries the synth DNS record
-	// pointing at K1's IP. ---
-	provider, err := agentMgr.CreateProvider(ctx, adminUserID, &agenttypes.Provider{
-		AccountID:   accountID,
-		ProviderID:  "openai_api",
-		Name:        "openai-test",
-		UpstreamURL: "https://api.openai.com",
-		APIKey:      "sk-test-key",
-		Enabled:     true,
-		Models:      []agenttypes.ProviderModel{{ID: "gpt-5.4"}},
-	}, clusterAddr)
-	require.NoError(t, err, "CreateProvider must succeed")
-
-	_, err = agentMgr.CreatePolicy(ctx, adminUserID, &agenttypes.Policy{
-		AccountID:              accountID,
-		Name:                   "p1",
-		Enabled:                true,
-		SourceGroups:           []string{groupAID},
-		DestinationProviderIDs: []string{provider.ID},
-	})
-	require.NoError(t, err, "CreatePolicy must succeed")
-
-	settings, err := am.Store.GetAgentNetworkSettings(ctx, store.LockingStrengthNone, accountID)
-	require.NoError(t, err)
-	fqdn := settings.Endpoint()
-
-	rdata1 := awaitZoneRData(clientCh, clusterAddr, fqdn, true)
-	require.Equal(t, proxyIP1, rdata1,
-		"client must receive a synth DNS record pointing at K1's overlay IP after the synth path runs")
-	drain(clientCh)
-
-	// --- Proxy restart: NEW keypair K2, same account, same cluster ---
-	require.NoError(t, peersMgr.CreateProxyPeer(ctx, accountID, proxyKey2, clusterAddr),
-		"second CreateProxyPeer (proxy restart with fresh keypair) must succeed")
-
-	peer2ID, err := am.Store.GetPeerIDByKey(ctx, store.LockingStrengthNone, proxyKey2)
-	require.NoError(t, err, "proxy peer for K2 must be persisted after restart")
-	require.NotEmpty(t, peer2ID)
-
-	require.NoError(t, am.MarkPeerConnected(ctx, proxyKey2, accountID, time.Now().UnixNano(), &types.NetworkMap{}),
-		"MarkPeerConnected for K2 must succeed")
-
-	// In production the agent's sync stream pulls a fresh NetworkMap as
-	// part of its normal reconcile cadence; in this isolated test
-	// MarkPeerConnected's affected-peer fan-out can race the channel-side
-	// buffer in a way that swallows the synth-DNS-bearing update before
-	// our await reads it. Trigger an explicit account-wide fan-out so the
-	// assertion below tests what production actually delivers, not the
-	// in-test buffer race.
-	am.UpdateAccountPeers(ctx, accountID, types.UpdateReason{Resource: types.UpdateResourcePeer, Operation: types.UpdateOperationUpdate})
-
-	account, err = am.Store.GetAccount(ctx, accountID)
-	require.NoError(t, err)
-	proxyIP2 := account.Peers[peer2ID].IP.String()
-	require.NotEmpty(t, proxyIP2, "K2 must have an assigned overlay IP")
-	require.NotEqual(t, proxyIP1, proxyIP2, "K2 must get a different overlay IP than K1 (sanity)")
-
-	// CRITICAL ASSERTION 1: K1 must no longer be in the store. The SqlStore
-	// returns ("", nil) for a missing key rather than NotFound, so assert
-	// on the returned ID being empty.
-	staleID, err := am.Store.GetPeerIDByKey(ctx, store.LockingStrengthNone, proxyKey1)
-	require.NoError(t, err, "GetPeerIDByKey for a missing peer must not error")
-	assert.Empty(t, staleID,
-		"stale embedded proxy peer K1 must be removed when a new embedded peer registers for the same (account, cluster); pre-fix this assertion fails because management never cleans up the prior peer record")
-
-	// CRITICAL ASSERTION 2: exactly one embedded proxy peer remains, and it
-	// is K2.
-	account, err = am.Store.GetAccount(ctx, accountID)
-	require.NoError(t, err)
-	embeddedKeys := []string{}
-	for _, p := range account.Peers {
-		if p.ProxyMeta.Embedded {
-			embeddedKeys = append(embeddedKeys, p.Key)
-		}
-	}
-	assert.Equal(t, []string{proxyKey2}, embeddedKeys,
-		"after a proxy restart exactly one embedded proxy peer should remain — the one with the new key K2")
-
-	// CRITICAL ASSERTION 3: the synth DNS record the client receives now
-	// points at K2's IP, not K1's.
-	rdata2 := awaitZoneRData(clientCh, clusterAddr, fqdn, true)
-	assert.Equal(t, proxyIP2, rdata2,
-		"after proxy restart, the client's synth DNS record must point at the NEW embedded peer's IP, not the stale K1 IP")
-}
--- a/management/server/job/manager.go
+++ b/management/server/job/manager.go
@@ -4,7 +4,6 @@ import (
 	"context"
 	"fmt"
 	"sync"
-	"time"

 	log "github.com/sirupsen/logrus"

@@ -21,11 +20,17 @@ type Event struct {
 	Response *proto.JobResponse
 }

+// PeerStream is the send side of a peer's Job stream. Sends are serialized by
+// its mutex; the receive side runs on the stream's gRPC handler goroutine.
+type PeerStream struct {
+	send func(*Event) error
+	mu   sync.Mutex
+}
+
 type Manager struct {
 	mu           *sync.RWMutex
-	jobChannels  map[string]*Channel // per-peer job streams
-	pending      map[string]*Event   // jobID → event
-	responseWait time.Duration
+	streams      map[string]*PeerStream // per-peer job streams
+	pending      map[string]*Event      // jobID → event
 	metrics      telemetry.AppMetrics
 	Store        store.Store
 	peersManager peers.Manager
@@ -34,9 +39,8 @@ type Manager struct {
 func NewJobManager(metrics telemetry.AppMetrics, store store.Store, peersManager peers.Manager) *Manager {

 	return &Manager{
-		jobChannels:  make(map[string]*Channel),
+		streams:      make(map[string]*PeerStream),
 		pending:      make(map[string]*Event),
-		responseWait: 5 * time.Minute,
 		metrics:      metrics,
 		mu:           &sync.RWMutex{},
 		Store:        store,
@@ -44,8 +48,9 @@ func NewJobManager(metrics telemetry.AppMetrics, store store.Store, peersManager
 	}
 }

-// CreateJobChannel creates or replaces a channel for a peer
-func (jm *Manager) CreateJobChannel(ctx context.Context, accountID, peerID string) *Channel {
+// RegisterStream registers the send side of a peer's Job stream, replacing any
+// previous registration for the peer.
+func (jm *Manager) RegisterStream(ctx context.Context, accountID, peerID string, send func(*Event) error) *PeerStream {
 	// all pending jobs stored in db for this peer should be failed
 	if err := jm.Store.MarkAllPendingJobsAsFailed(ctx, accountID, peerID, "Pending job cleanup: marked as failed automatically due to being stuck too long"); err != nil {
 		log.WithContext(ctx).Error(err.Error())
@@ -54,23 +59,41 @@ func (jm *Manager) CreateJobChannel(ctx context.Context, accountID, peerID strin
 	jm.mu.Lock()
 	defer jm.mu.Unlock()

-	if ch, ok := jm.jobChannels[peerID]; ok {
-		ch.Close()
-		delete(jm.jobChannels, peerID)
-	}
+	stream := &PeerStream{send: send}
+	jm.streams[peerID] = stream
+	return stream
+}

-	ch := NewChannel()
-	jm.jobChannels[peerID] = ch
-	return ch
+// UnregisterStream removes a peer's stream registration and fails its pending
+// jobs. It is a no-op if the registration was already replaced by a newer
+// stream of the same peer.
+func (jm *Manager) UnregisterStream(ctx context.Context, accountID, peerID string, stream *PeerStream) {
+	jm.mu.Lock()
+	defer jm.mu.Unlock()
+
+	if jm.streams[peerID] != stream {
+		return
+	}
+	delete(jm.streams, peerID)
+
+	for jobID, ev := range jm.pending {
+		if ev.PeerID == peerID {
+			// if the client disconnect and there is pending job then mark it as failed
+			if err := jm.Store.MarkPendingJobsAsFailed(ctx, accountID, peerID, jobID, "Time out peer disconnected"); err != nil {
+				log.WithContext(ctx).Errorf("failed to mark pending jobs as failed: %v", err)
+			}
+			delete(jm.pending, jobID)
+		}
+	}
 }

 // SendJob sends a job to a peer and tracks it as pending
 func (jm *Manager) SendJob(ctx context.Context, accountID, peerID string, req *proto.JobRequest) error {
 	jm.mu.RLock()
-	ch, ok := jm.jobChannels[peerID]
+	stream, ok := jm.streams[peerID]
 	jm.mu.RUnlock()
 	if !ok {
-		return fmt.Errorf("peer %s has no channel", peerID)
+		return fmt.Errorf("peer %s has no stream", peerID)
 	}

 	event := &Event{
@@ -82,7 +105,10 @@ func (jm *Manager) SendJob(ctx context.Context, accountID, peerID string, req *p
 	jm.pending[string(req.ID)] = event
 	jm.mu.Unlock()

-	if err := ch.AddEvent(ctx, jm.responseWait, event); err != nil {
+	stream.mu.Lock()
+	err := stream.send(event)
+	stream.mu.Unlock()
+	if err != nil {
 		jm.cleanup(ctx, accountID, string(req.ID), err.Error())
 		return err
 	}
@@ -127,27 +153,6 @@ func (jm *Manager) HandleResponse(ctx context.Context, resp *proto.JobResponse,
 	return nil
 }

-// CloseChannel closes a peer’s channel and cleans up its jobs
-func (jm *Manager) CloseChannel(ctx context.Context, accountID, peerID string) {
-	jm.mu.Lock()
-	defer jm.mu.Unlock()
-
-	if ch, ok := jm.jobChannels[peerID]; ok {
-		ch.Close()
-		delete(jm.jobChannels, peerID)
-	}
-
-	for jobID, ev := range jm.pending {
-		if ev.PeerID == peerID {
-			// if the client disconnect and there is pending job then mark it as failed
-			if err := jm.Store.MarkPendingJobsAsFailed(ctx, accountID, peerID, jobID, "Time out peer disconnected"); err != nil {
-				log.WithContext(ctx).Errorf("failed to mark pending jobs as failed: %v", err)
-			}
-			delete(jm.pending, jobID)
-		}
-	}
-}
-
 // cleanup removes a pending job safely
 func (jm *Manager) cleanup(ctx context.Context, accountID, jobID string, reason string) {
 	jm.mu.Lock()
@@ -165,7 +170,7 @@ func (jm *Manager) IsPeerConnected(peerID string) bool {
 	jm.mu.RLock()
 	defer jm.mu.RUnlock()

-	_, ok := jm.jobChannels[peerID]
+	_, ok := jm.streams[peerID]
 	return ok
 }

--- a/management/server/job/manager_test.go
+++ b/management/server/job/manager_test.go
@@ -0,0 +1,90 @@
+package job
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/golang/mock/gomock"
+	"github.com/stretchr/testify/require"
+
+	"github.com/netbirdio/netbird/management/server/store"
+	"github.com/netbirdio/netbird/shared/management/proto"
+)
+
+func newTestManager(t *testing.T) (*Manager, *store.MockStore) {
+	t.Helper()
+	ctrl := gomock.NewController(t)
+	t.Cleanup(ctrl.Finish)
+	mockStore := store.NewMockStore(ctrl)
+	return NewJobManager(nil, mockStore, nil), mockStore
+}
+
+func TestSendJobDeliversThroughRegisteredStream(t *testing.T) {
+	ctx := context.Background()
+	manager, mockStore := newTestManager(t)
+	mockStore.EXPECT().MarkAllPendingJobsAsFailed(gomock.Any(), "acc", "peer1", gomock.Any()).Return(nil)
+
+	var sent []*Event
+	manager.RegisterStream(ctx, "acc", "peer1", func(event *Event) error {
+		sent = append(sent, event)
+		return nil
+	})
+	require.True(t, manager.IsPeerConnected("peer1"))
+
+	err := manager.SendJob(ctx, "acc", "peer1", &proto.JobRequest{ID: []byte("job1")})
+	require.NoError(t, err)
+	require.Len(t, sent, 1)
+	require.Equal(t, "peer1", sent[0].PeerID)
+	require.True(t, manager.IsPeerHasPendingJobs("peer1"))
+}
+
+func TestSendJobWithoutStream(t *testing.T) {
+	manager, _ := newTestManager(t)
+	err := manager.SendJob(context.Background(), "acc", "peer1", &proto.JobRequest{ID: []byte("job1")})
+	require.Error(t, err)
+}
+
+func TestSendJobFailureCleansPending(t *testing.T) {
+	ctx := context.Background()
+	manager, mockStore := newTestManager(t)
+	mockStore.EXPECT().MarkAllPendingJobsAsFailed(gomock.Any(), "acc", "peer1", gomock.Any()).Return(nil)
+	mockStore.EXPECT().MarkPendingJobsAsFailed(gomock.Any(), "acc", "peer1", "job1", gomock.Any()).Return(nil)
+
+	manager.RegisterStream(ctx, "acc", "peer1", func(*Event) error {
+		return errors.New("stream broken")
+	})
+
+	err := manager.SendJob(ctx, "acc", "peer1", &proto.JobRequest{ID: []byte("job1")})
+	require.Error(t, err)
+	require.False(t, manager.IsPeerHasPendingJobs("peer1"))
+}
+
+func TestUnregisterStreamIgnoresSupersededRegistration(t *testing.T) {
+	ctx := context.Background()
+	manager, mockStore := newTestManager(t)
+	mockStore.EXPECT().MarkAllPendingJobsAsFailed(gomock.Any(), "acc", "peer1", gomock.Any()).Return(nil).Times(2)
+
+	first := manager.RegisterStream(ctx, "acc", "peer1", func(*Event) error { return nil })
+	second := manager.RegisterStream(ctx, "acc", "peer1", func(*Event) error { return nil })
+
+	manager.UnregisterStream(ctx, "acc", "peer1", first)
+	require.True(t, manager.IsPeerConnected("peer1"), "stale unregister must not remove the replacement stream")
+
+	manager.UnregisterStream(ctx, "acc", "peer1", second)
+	require.False(t, manager.IsPeerConnected("peer1"))
+}
+
+func TestUnregisterStreamFailsPendingJobs(t *testing.T) {
+	ctx := context.Background()
+	manager, mockStore := newTestManager(t)
+	mockStore.EXPECT().MarkAllPendingJobsAsFailed(gomock.Any(), "acc", "peer1", gomock.Any()).Return(nil)
+	mockStore.EXPECT().MarkPendingJobsAsFailed(gomock.Any(), "acc", "peer1", "job1", gomock.Any()).Return(nil)
+
+	stream := manager.RegisterStream(ctx, "acc", "peer1", func(*Event) error { return nil })
+	require.NoError(t, manager.SendJob(ctx, "acc", "peer1", &proto.JobRequest{ID: []byte("job1")}))
+	require.True(t, manager.IsPeerHasPendingJobs("peer1"))
+
+	manager.UnregisterStream(ctx, "acc", "peer1", stream)
+	require.False(t, manager.IsPeerHasPendingJobs("peer1"))
+}
--- a/management/server/types/networkmap_components.go
+++ b/management/server/types/networkmap_components.go
@@ -7,6 +7,7 @@ import (
 	"slices"
 	"strconv"
 	"strings"
+	"sync"
 	"time"

 	"github.com/netbirdio/netbird/client/ssh/auth"
@@ -42,6 +43,14 @@ type NetworkMapComponents struct {
 	PostureFailedPeers map[string]map[string]struct{}

 	RouterPeers map[string]*nbpeer.Peer
+
+	routesByPeerOnce sync.Once
+	routesByPeerIdx  map[string][]routeIndexEntry
+}
+
+type routeIndexEntry struct {
+	route    *route.Route
+	viaGroup bool
 }

 type AccountSettingsInfo struct {
@@ -530,33 +539,43 @@ func (c *NetworkMapComponents) getRoutingPeerRoutes(peerID string) (enabledRoute
 		disabledRoutes = append(disabledRoutes, r)
 	}

-	for _, r := range c.Routes {
-		for _, groupID := range r.PeerGroups {
-			group := c.GetGroupInfo(groupID)
-			if group == nil {
-				continue
-			}
-			for _, id := range group.Peers {
-				if id != peerID {
-					continue
-				}
-
-				newPeerRoute := r.Copy()
-				newPeerRoute.Peer = id
-				newPeerRoute.PeerGroups = nil
-				newPeerRoute.ID = route.ID(string(r.ID) + ":" + id)
-				takeRoute(newPeerRoute)
-				break
-			}
-		}
-		if r.Peer == peerID {
-			takeRoute(r.Copy())
+	for _, entry := range c.routesByPeer()[peerID] {
+		if entry.viaGroup {
+			newPeerRoute := entry.route.Copy()
+			newPeerRoute.PeerGroups = nil
+			newPeerRoute.ID = route.ID(string(entry.route.ID) + ":" + peerID)
+			takeRoute(newPeerRoute)
+			continue
 		}
+		takeRoute(entry.route.Copy())
 	}

 	return enabledRoutes, disabledRoutes
 }

+func (c *NetworkMapComponents) routesByPeer() map[string][]routeIndexEntry {
+	c.routesByPeerOnce.Do(func() {
+		idx := make(map[string][]routeIndexEntry)
+		for _, r := range c.Routes {
+			for _, groupID := range r.PeerGroups {
+				group := c.GetGroupInfo(groupID)
+				if group == nil {
+					continue
+				}
+				for _, id := range group.Peers {
+					idx[id] = append(idx[id], routeIndexEntry{route: r, viaGroup: true})
+				}
+			}
+			if r.Peer != "" {
+				idx[r.Peer] = append(idx[r.Peer], routeIndexEntry{route: r})
+			}
+		}
+		c.routesByPeerIdx = idx
+	})
+
+	return c.routesByPeerIdx
+}
+
 func (c *NetworkMapComponents) filterRoutesByGroups(routes []*route.Route, groupListMap LookupMap) []*route.Route {
 	var filteredRoutes []*route.Route
 	for _, r := range routes {
--- a/proxy/internal/middleware/builtin/llm_guardrail/middleware.go
+++ b/proxy/internal/middleware/builtin/llm_guardrail/middleware.go
@@ -25,6 +25,14 @@ const (
 	denyCodeModel    = "llm_policy.model_blocked"
 	denyReasonModel  = "model_blocked"
 	denyMessageModel = "model is not in the policy allowlist"
+	// Deny reason used when an allowlist is configured but the request model
+	// could not be determined. URL/path-routed providers (AWS Bedrock, Google
+	// Vertex, ...) carry the model outside the JSON body, so a request shape the
+	// parser does not recognise reaches the guardrail with no model. Such a
+	// request must be denied (fail closed), never waved through.
+	denyCodeModelUnknown    = "llm_policy.model_unknown"
+	denyReasonModelUnknown  = "model_unknown"
+	denyMessageModelUnknown = "request model could not be determined for the policy allowlist"
 )

 // Middleware enforces the model allowlist and optionally captures the
@@ -108,23 +116,37 @@ func (m *Middleware) evaluateAllowlist(model string, modelPresent bool) *middlew
 	if len(m.cfg.ModelAllowlist) == 0 {
 		return nil
 	}
-	if !modelPresent {
-		return nil
+	// Fail closed: with an allowlist configured, a request whose model the
+	// upstream parser could not extract (absent or empty) must be denied rather
+	// than allowed. This is what enforces the allowlist for URL/path-routed
+	// providers (Bedrock, Vertex, ...) whose model lives outside the JSON body.
+	if !modelPresent || normaliseModel(model) == "" {
+		return denyModel("", denyCodeModelUnknown, denyMessageModelUnknown, denyReasonModelUnknown)
 	}
 	if m.modelInAllowlist(model) {
 		return nil
 	}
+	return denyModel(model, denyCodeModel, denyMessageModel, denyReasonModel)
+}
+
+// denyModel builds a 403 deny Output for a model-allowlist rejection. model is
+// included in the details only when non-empty.
+func denyModel(model, code, message, reason string) *middleware.Output {
+	details := map[string]string{}
+	if model != "" {
+		details["model"] = model
+	}
 	return &middleware.Output{
 		Decision:   middleware.DecisionDeny,
 		DenyStatus: 403,
 		DenyReason: &middleware.DenyReason{
-			Code:    denyCodeModel,
-			Message: denyMessageModel,
-			Details: map[string]string{"model": model},
+			Code:    code,
+			Message: message,
+			Details: details,
 		},
 		Metadata: []middleware.KV{
 			{Key: middleware.KeyLLMPolicyDecision, Value: "deny"},
-			{Key: middleware.KeyLLMPolicyReason, Value: denyReasonModel},
+			{Key: middleware.KeyLLMPolicyReason, Value: reason},
 		},
 	}
 }
--- a/proxy/internal/middleware/builtin/llm_guardrail/middleware_test.go
+++ b/proxy/internal/middleware/builtin/llm_guardrail/middleware_test.go
@@ -102,13 +102,44 @@ func TestAllowlistCaseInsensitive(t *testing.T) {
 	}
 }

-func TestAllowlistMissingModelKeyAllows(t *testing.T) {
+func TestAllowlistMissingModelKeyDenies(t *testing.T) {
+	// Fail closed: with an allowlist configured, a request whose model the
+	// parser could not extract (URL/path-routed providers such as Bedrock or
+	// Vertex whose shape wasn't recognised) must be denied, not allowed.
 	mw := New(Config{ModelAllowlist: []string{"gpt-4o"}})
 	out, err := mw.Invoke(context.Background(), newInput())
 	require.NoError(t, err)
-	assert.Equal(t, middleware.DecisionAllow, out.Decision, "missing model key must allow even with non-empty allowlist")
+	require.NotNil(t, out)
+	assert.Equal(t, middleware.DecisionDeny, out.Decision, "absent model must be denied when an allowlist is set")
+	assert.Equal(t, 403, out.DenyStatus, "deny status must be 403")
+	require.NotNil(t, out.DenyReason, "deny reason must be populated")
+	assert.Equal(t, "llm_policy.model_unknown", out.DenyReason.Code, "deny code must be model_unknown")
 	dec, _ := metaValue(t, out.Metadata, middleware.KeyLLMPolicyDecision)
-	assert.Equal(t, "allow", dec, "decision must be allow when model key is absent")
+	assert.Equal(t, "deny", dec, "decision must be deny when model key is absent")
+	reason, _ := metaValue(t, out.Metadata, middleware.KeyLLMPolicyReason)
+	assert.Equal(t, "model_unknown", reason, "reason metadata must be model_unknown")
+}
+
+func TestAllowlistEmptyModelValueDenies(t *testing.T) {
+	// A present-but-empty model is as undeterminable as an absent one.
+	mw := New(Config{ModelAllowlist: []string{"gpt-4o"}})
+	out, err := mw.Invoke(context.Background(), newInput(
+		middleware.KV{Key: middleware.KeyLLMModel, Value: "   "},
+	))
+	require.NoError(t, err)
+	require.NotNil(t, out)
+	assert.Equal(t, middleware.DecisionDeny, out.Decision, "empty model must be denied when an allowlist is set")
+	require.NotNil(t, out.DenyReason, "deny reason must be populated")
+	assert.Equal(t, "llm_policy.model_unknown", out.DenyReason.Code, "deny code must be model_unknown")
+}
+
+func TestAllowlistEmptyListAllowsMissingModel(t *testing.T) {
+	// Without an allowlist there is nothing to enforce, so a missing model is
+	// still allowed — the fail-closed rule only applies when a list is set.
+	mw := New(Config{})
+	out, err := mw.Invoke(context.Background(), newInput())
+	require.NoError(t, err)
+	assert.Equal(t, middleware.DecisionAllow, out.Decision, "no allowlist must allow even without a model")
 }

 func TestPromptCaptureDisabledEmitsNoPrompt(t *testing.T) {
--- a/proxy/internal/middleware/builtin/llm_request_parser/guardrail_allowlist_test.go
+++ b/proxy/internal/middleware/builtin/llm_request_parser/guardrail_allowlist_test.go
@@ -0,0 +1,106 @@
+package llm_request_parser
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/netbirdio/netbird/proxy/internal/middleware"
+	"github.com/netbirdio/netbird/proxy/internal/middleware/builtin/llm_guardrail"
+)
+
+// runParserGuardrail runs the request parser then the model-allowlist guardrail
+// in SlotOnRequest order, threading the parser's metadata into the guardrail the
+// same way the real chain does. It returns the guardrail decision so tests can
+// assert allowlist enforcement for URL/path-routed providers end to end.
+func runParserGuardrail(t *testing.T, url string, body []byte, allowlist []string) *middleware.Output {
+	t.Helper()
+	parser := newMiddleware(t)
+	parsed, err := parser.Invoke(context.Background(), &middleware.Input{
+		Slot: middleware.SlotOnRequest,
+		URL:  url,
+		Body: body,
+	})
+	require.NoError(t, err, "parser must not error")
+
+	guard := llm_guardrail.New(llm_guardrail.Config{ModelAllowlist: allowlist})
+	out, err := guard.Invoke(context.Background(), &middleware.Input{
+		Slot:     middleware.SlotOnRequest,
+		Metadata: parsed.Metadata,
+	})
+	require.NoError(t, err, "guardrail must not error")
+	require.NotNil(t, out, "guardrail must return an output")
+	return out
+}
+
+// TestModelAllowlist_URLRoutedProviders validates that the model allowlist is
+// enforced for providers whose model travels in the URL path (AWS Bedrock,
+// Google Vertex) rather than the JSON body. The "unknown action" case is the
+// regression guard for #6751: a Bedrock request shape the parser cannot map to a
+// model must fail closed under an allowlist instead of bypassing it.
+func TestModelAllowlist_URLRoutedProviders(t *testing.T) {
+	const bedrockBody = `{"anthropic_version":"bedrock-2023-05-31","messages":[{"role":"user","content":"hi"}]}`
+	const vertexBody = `{"anthropic_version":"vertex-2023-10-16","messages":[{"role":"user","content":"hi"}]}`
+
+	tests := []struct {
+		name      string
+		url       string
+		body      string
+		allowlist []string
+		decision  middleware.Decision
+		denyCode  string
+	}{
+		{
+			name:      "bedrock allowed model passes",
+			url:       "https://bedrock-runtime.us-east-1.amazonaws.com/model/us.anthropic.claude-haiku-4-5-v1:0/invoke",
+			body:      bedrockBody,
+			allowlist: []string{"anthropic.claude-haiku-4-5"},
+			decision:  middleware.DecisionAllow,
+		},
+		{
+			name:      "bedrock disallowed model denied",
+			url:       "https://bedrock-runtime.us-east-1.amazonaws.com/model/us.anthropic.claude-opus-4-8-v1:0/invoke",
+			body:      bedrockBody,
+			allowlist: []string{"anthropic.claude-haiku-4-5"},
+			decision:  middleware.DecisionDeny,
+			denyCode:  "llm_policy.model_blocked",
+		},
+		{
+			name:      "bedrock unknown action fails closed",
+			url:       "https://bedrock-runtime.us-east-1.amazonaws.com/model/us.anthropic.claude-opus-4-8-v1:0/some-future-action",
+			body:      bedrockBody,
+			allowlist: []string{"anthropic.claude-haiku-4-5"},
+			decision:  middleware.DecisionDeny,
+			denyCode:  "llm_policy.model_unknown",
+		},
+		{
+			name:      "vertex disallowed model denied",
+			url:       "/v1/projects/p/locations/global/publishers/anthropic/models/claude-opus-4-8@20250101:rawPredict",
+			body:      vertexBody,
+			allowlist: []string{"claude-haiku-4-5"},
+			decision:  middleware.DecisionDeny,
+			denyCode:  "llm_policy.model_blocked",
+		},
+		{
+			name:      "vertex allowed model passes",
+			url:       "/v1/projects/p/locations/global/publishers/anthropic/models/claude-haiku-4-5@20250101:rawPredict",
+			body:      vertexBody,
+			allowlist: []string{"claude-haiku-4-5"},
+			decision:  middleware.DecisionAllow,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			out := runParserGuardrail(t, tt.url, []byte(tt.body), tt.allowlist)
+			assert.Equal(t, tt.decision, out.Decision, "unexpected decision for %s", tt.name)
+			if tt.decision == middleware.DecisionDeny {
+				require.NotNil(t, out.DenyReason, "deny reason must be set for %s", tt.name)
+				assert.Equal(t, 403, out.DenyStatus, "deny status must be 403 for %s", tt.name)
+				assert.Equal(t, tt.denyCode, out.DenyReason.Code, "deny code for %s", tt.name)
+			}
+		})
+	}
+}
--- a/proxy/internal/roundtrip/netbird.go
+++ b/proxy/internal/roundtrip/netbird.go
@@ -30,12 +30,6 @@ import (

 const deviceNamePrefix = "ingress-proxy-"

-// envProxyRosenpass toggles Rosenpass (permissive) on the embedded proxy client. Defaults to on.
-const envProxyRosenpass = "NB_PROXY_ROSENPASS" //nolint:gosec // env var name, not a credential
-
-// envProxyClientLogLevel sets the embedded NetBird client's log level.
-const envProxyClientLogLevel = "NB_PROXY_CLIENT_LOG_LEVEL"
-
 const clientStopTimeout = 30 * time.Second

 const createProxyPeerTimeout = 30 * time.Second
@@ -359,11 +353,11 @@ func (n *NetBird) createClientEntry(ctx context.Context, accountID types.Account
 	// NB_PROXY_CLIENT_LOG_LEVEL (e.g. "trace") to surface the embedded NetBird
 	// client's relay / signal / handshake detail for local debugging.
 	clientLogLevel := log.WarnLevel.String()
-	if v := strings.TrimSpace(os.Getenv(envProxyClientLogLevel)); v != "" {
+	if v := strings.TrimSpace(os.Getenv("NB_PROXY_CLIENT_LOG_LEVEL")); v != "" {
 		if lvl, err := log.ParseLevel(v); err == nil {
 			clientLogLevel = lvl.String()
 		} else {
-			n.logger.Warnf("invalid %s %q, using %q: %v", envProxyClientLogLevel, v, clientLogLevel, err)
+			n.logger.Warnf("invalid NB_PROXY_CLIENT_LOG_LEVEL %q, using %q: %v", v, clientLogLevel, err)
 		}
 	}

@@ -373,26 +367,15 @@ func (n *NetBird) createClientEntry(ctx context.Context, accountID types.Account
 		}
 	})

-	// Rosenpass runs in permissive mode by default so the embedded proxy can
-	// establish connections with Rosenpass-enabled peers (which otherwise fail
-	// on a PSK mismatch) while still falling back to plain WireGuard for peers
-	// that do not run Rosenpass. Set NB_PROXY_ROSENPASS=false to disable it.
-	rosenpassEnabled := true
-	if v, ok := envBool(envProxyRosenpass, n.logger); ok {
-		rosenpassEnabled = v
-	}
-
 	// Create embedded NetBird client with the generated private key.
 	// The peer has already been created via CreateProxyPeer RPC with the public key.
 	wgPort := int(n.clientCfg.WGPort)
 	embedOpts := embed.Options{
-		DeviceName:          deviceNamePrefix + n.proxyID,
-		ManagementURL:       n.clientCfg.MgmtAddr,
-		PrivateKey:          privateKey.String(),
-		LogLevel:            clientLogLevel,
-		BlockInbound:        n.clientCfg.BlockInbound,
-		EnableRosenpass:     rosenpassEnabled,
-		RosenpassPermissive: rosenpassEnabled,
+		DeviceName:    deviceNamePrefix + n.proxyID,
+		ManagementURL: n.clientCfg.MgmtAddr,
+		PrivateKey:    privateKey.String(),
+		LogLevel:      clientLogLevel,
+		BlockInbound:  n.clientCfg.BlockInbound,
 		// The embedded proxy peer must never be a stepping stone into
 		// the proxy host's LAN: it only exists to reach NetBird mesh
 		// targets or, when direct_upstream is set, the host network
@@ -916,8 +899,6 @@ func logEmbedOptions(logger *log.Logger, accountID types.AccountID, serviceID ty
 		"mtu":                   mtu,
 		"block_inbound":         opts.BlockInbound,
 		"block_lan_access":      opts.BlockLANAccess,
-		"rosenpass_enabled":     opts.EnableRosenpass,
-		"rosenpass_permissive":  opts.RosenpassPermissive,
 		"disable_ipv6":          opts.DisableIPv6,
 		"disable_client_routes": opts.DisableClientRoutes,
 		"no_userspace":          opts.NoUserspace,
--- a/shared/management/proto/management.pb.go
+++ b/shared/management/proto/management.pb.go
--- a/shared/management/proto/management.proto
+++ b/shared/management/proto/management.proto
@@ -486,22 +486,6 @@ message RemotePeerConfig {
  string fqdn = 4;

  string agentVersion = 5;
-
-  // lazyState is the management per-peer override for lazy (on-demand)
-  // connections to this remote peer. LazyStateDefault follows the account-wide
-  // flag; LazyStateLazy forces lazy; LazyStateEager forces an always-active
-  // connection. A local NB_LAZY_CONN/MDM override still wins over this.
-  LazyState lazyState = 6;
-}
-
-// LazyState is the management per-peer override for lazy connections.
-enum LazyState {
-  // Follow the account-wide lazy connection flag.
-  LazyStateDefault = 0;
-  // Force a lazy (on-demand) connection regardless of the account flag.
-  LazyStateLazy = 1;
-  // Force an always-active connection regardless of the account flag.
-  LazyStateEager = 2;
 }

 // SSHConfig represents SSH configurations of a peer.
Author	SHA1	Message	Date
pascal	8fd098dcc7	fix linter	2026-07-16 13:24:34 +02:00
pascal	820303c71d	fix interval and job cancel	2026-07-16 02:13:51 +02:00
pascal	ad31406494	improve jobs endpoint	2026-07-16 02:07:15 +02:00
pascal	4d8d0e30db	use native grpc + unify token refresh	2026-07-16 01:22:37 +02:00
pascal	5aa2a748d7	use native grpc + unify token refresh	2026-07-16 01:22:30 +02:00
Pascal Fischer	e1a24376ab	[management] build routes for peer cache on network map components (#6780 )	2026-07-15 18:24:48 +02:00
Pascal Fischer	8f901f8899	[management] enable pprof via env var (#6778 )	2026-07-15 12:05:40 +02:00
Maycon Santos	c6bf5fbbfb	[management,client] 0.74.5 branch sync (#6769 ) ## Describe your changes * [proxy] enforce model allowlist for URL-routed providers (Bedrock/Vertex) by @mlsmaycon in https://github.com/netbirdio/netbird/pull/6764 * [management] Remove proxy peer stale deduplication logic by @mlsmaycon in https://github.com/netbirdio/netbird/pull/6768 ## Issue ticket number and link ## Stack <!-- branch-stack --> ### Checklist - [ ] Is it a bug fix - [ ] Is a typo/documentation fix - [ ] Is a feature enhancement - [ ] It is a refactor - [ ] Created tests that fail without the change (if possible) - [ ] This change does not modify the public API, gRPC protocols, functionality behavior, CLI / service flags, or introduce a new feature — OR I have discussed it with the NetBird team beforehand (link the issue / Slack thread in the description). See [CONTRIBUTING.md](https://github.com/netbirdio/netbird/blob/main/CONTRIBUTING.md#discuss-changes-with-the-netbird-team-first). > By submitting this pull request, you confirm that you have read and agree to the terms of the [Contributor License Agreement](https://github.com/netbirdio/netbird/blob/main/CONTRIBUTOR_LICENSE_AGREEMENT.md). ## Documentation Select exactly one: - [ ] I added/updated documentation for this change - [x] Documentation is not needed for this change (explain why) ### Docs PR URL (required if "docs added" is checked) Paste the PR link from https://github.com/netbirdio/docs here: https://github.com/netbirdio/docs/pull/__ <!-- This is an auto-generated comment: release notes by coderabbit.ai --> ## Summary by CodeRabbit - New Features - Added model-allowlist guardrails for path-routed providers, including Bedrock and Vertex. - Added Bedrock request support for chat interactions. - Added guardrail management capabilities. - Bug Fixes - Requests with missing or blank model identifiers are now denied when a model allowlist is configured, improving fail-closed protection. - Corrected provider-specific request handling and session tracking for Bedrock interactions. - Tests - Expanded coverage for allowlist enforcement and provider routing scenarios. <!-- end of auto-generated comment: release notes by coderabbit.ai --> --------- Co-authored-by: Theodor Midtlien <theodor@midtlien.com> Co-authored-by: blaugrau90 <61945343+blaugrau90@users.noreply.github.com> Co-authored-by: Viktor Liu <17948409+lixmal@users.noreply.github.com>	2026-07-14 21:22:40 +02:00
David Fry	e70a69bbcf	[client] Restore residual state in foreground mode before login (#6707 ) * Improved residual state restoration during foreground startup and foreground login, ensuring consistent recovery with stale states. * Foreground flows now initialize advanced routing so stale routes are bypassed during login.	2026-07-14 17:43:59 +02:00