diff --git a/test-integ/upgrade/usage_profiles/sd_wan_segments_test.go b/test-integ/upgrade/usage_profiles/sd_wan_segments_test.go
new file mode 100644
index 0000000000..98a0873c17
--- /dev/null
+++ b/test-integ/upgrade/usage_profiles/sd_wan_segments_test.go
@@ -0,0 +1,70 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: BUSL-1.1
+
+package usage_profiles
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/hashicorp/consul/test/integration/consul-container/libs/utils"
+	"github.com/hashicorp/consul/testing/deployer/sprawl"
+	"github.com/hashicorp/consul/testing/deployer/sprawl/sprawltest"
+	"github.com/hashicorp/consul/testing/deployer/topology"
+)
+
+const (
+	// The long term support version
+	ltsVersion = "1.15.7"
+)
+
+// Test_Upgrade_ServiceDiscovery_Wan_Segment test upgrade from a source version
+// to a specified long term support version
+// Clusters: multi-segment and multi-cluster (TODO)
+// Workload: service discovery (no mesh) (TODO)
+func Test_Upgrade_ServiceDiscovery_Wan_Segment(t *testing.T) {
+	utils.LatestVersion = "1.10.8"
+	utils.TargetVersion = ltsVersion
+
+	dc1, err := createTopology("dc1")
+	require.NoError(t, err)
+	t.Log("Created topology:", dc1.Name, "enterprise:", utils.IsEnterprise())
+
+	toplogyConfig := &topology.Config{
+		Networks: []*topology.Network{
+			{Name: "dc1"},
+		},
+	}
+	toplogyConfig.Clusters = append(toplogyConfig.Clusters, dc1)
+	sp := sprawltest.Launch(t, toplogyConfig)
+
+	cfg := sp.Config()
+	require.NoError(t, sp.Upgrade(cfg, "dc1", sprawl.UpgradeTypeStandard, utils.TargetImages(), nil))
+	t.Log("Finished standard upgrade ...")
+
+	time.Sleep(30 * time.Second)
+}
+
+func createTopology(name string) (*topology.Cluster, error) {
+	clu := &topology.Cluster{
+		Name:   name,
+		Images: utils.LatestImages(),
+		Nodes: []*topology.Node{
+			{
+				Kind: topology.NodeKindServer,
+				Name: "dc1-server1",
+				Addresses: []*topology.Address{
+					{Network: "dc1"},
+				},
+			},
+			{
+				Kind: topology.NodeKindClient,
+				Name: "dc1-client1",
+			},
+		},
+		Enterprise: utils.IsEnterprise(),
+	}
+	return clu, nil
+}
diff --git a/testing/deployer/go.mod b/testing/deployer/go.mod
index dbec818515..a78df479dd 100644
--- a/testing/deployer/go.mod
+++ b/testing/deployer/go.mod
@@ -13,6 +13,7 @@ require (
 	github.com/hashicorp/go-hclog v1.5.0
 	github.com/hashicorp/go-multierror v1.1.1
 	github.com/hashicorp/go-rootcerts v1.0.2
+	github.com/hashicorp/go-version v1.2.1
 	github.com/hashicorp/hcl/v2 v2.16.2
 	github.com/mitchellh/copystructure v1.2.0
 	github.com/rboyer/safeio v0.2.2
@@ -40,7 +41,6 @@ require (
 	github.com/hashicorp/go-immutable-radix v1.3.1 // indirect
 	github.com/hashicorp/go-netaddrs v0.1.0 // indirect
 	github.com/hashicorp/go-uuid v1.0.3 // indirect
-	github.com/hashicorp/go-version v1.2.1 // indirect
 	github.com/hashicorp/golang-lru v0.5.4 // indirect
 	github.com/hashicorp/serf v0.10.1 // indirect
 	github.com/json-iterator/go v1.1.12 // indirect
diff --git a/testing/deployer/sprawl/acl.go b/testing/deployer/sprawl/acl.go
index 99179ac586..92f4701840 100644
--- a/testing/deployer/sprawl/acl.go
+++ b/testing/deployer/sprawl/acl.go
@@ -146,6 +146,56 @@ func (s *Sprawl) createAnonymousPolicy(cluster *topology.Cluster) error {
 	return nil
 }
 
+// assignAgentJoinPolicyToAnonymousToken is used only for version prior to agent token
+func (s *Sprawl) assignAgentJoinPolicyToAnonymousToken(cluster *topology.Cluster) error {
+	var (
+		client = s.clients[cluster.Name]
+	)
+
+	acl := client.ACL()
+	anonymousTok, _, err := acl.TokenRead(anonymousTokenAccessorID, &api.QueryOptions{})
+	if err != nil {
+		return nil
+	}
+
+	rule := `
+service_prefix "" {
+	policy = "read"
+}
+	
+agent_prefix "" {
+	policy = "read"
+}
+	
+node_prefix "" {
+	policy = "write"
+}
+`
+	policy, _, err := acl.PolicyCreate(
+		&api.ACLPolicy{
+			Name:  "client-join-policy",
+			Rules: rule,
+		},
+		&api.WriteOptions{},
+	)
+
+	if err != nil {
+		return err
+	}
+
+	anonymousTok.Policies = append(anonymousTok.Policies,
+		&api.ACLLink{
+			Name: policy.Name,
+		},
+	)
+	_, _, err = acl.TokenUpdate(anonymousTok, &api.WriteOptions{})
+	if err != nil {
+		return nil
+	}
+
+	return nil
+}
+
 func (s *Sprawl) createAgentTokens(cluster *topology.Cluster) error {
 	var (
 		client = s.clients[cluster.Name]
@@ -158,18 +208,20 @@ func (s *Sprawl) createAgentTokens(cluster *topology.Cluster) error {
 			continue
 		}
 
-		if tok := s.secrets.ReadAgentToken(cluster.Name, node.ID()); tok == "" {
-			token, err := CreateOrUpdateToken(client, tokenForNode(node, cluster.Enterprise))
-			if err != nil {
-				return err
+		if node.Images.GreaterThanVersion(topology.MinVersionAgentTokenPartition) {
+			if tok := s.secrets.ReadAgentToken(cluster.Name, node.ID()); tok == "" {
+				token, err := CreateOrUpdateToken(client, tokenForNode(node, cluster.Enterprise))
+				if err != nil {
+					return fmt.Errorf("node %s: %w", node.Name, err)
+				}
+
+				logger.Debug("created agent token",
+					"node", node.ID(),
+					"token", token.SecretID,
+				)
+
+				s.secrets.SaveAgentToken(cluster.Name, node.ID(), token.SecretID)
 			}
-
-			logger.Debug("created agent token",
-				"node", node.ID(),
-				"token", token.SecretID,
-			)
-
-			s.secrets.SaveAgentToken(cluster.Name, node.ID(), token.SecretID)
 		}
 	}
 
diff --git a/testing/deployer/sprawl/boot.go b/testing/deployer/sprawl/boot.go
index b8b123482e..38407d50ae 100644
--- a/testing/deployer/sprawl/boot.go
+++ b/testing/deployer/sprawl/boot.go
@@ -266,7 +266,7 @@ func (s *Sprawl) initConsulServers() error {
 		s.waitForLocalWrites(cluster, mgmtToken)
 
 		// Create tenancies so that the ACL tokens and clients have somewhere to go.
-		if cluster.Enterprise {
+		if cluster.Enterprise && node.Images.GreaterThanVersion(topology.MinVersionAgentTokenPartition) {
 			if err := s.initTenancies(cluster); err != nil {
 				return fmt.Errorf("initTenancies[%s]: %w", cluster.Name, err)
 			}
@@ -287,12 +287,19 @@ func (s *Sprawl) initConsulServers() error {
 			return fmt.Errorf("createAnonymousToken[%s]: %w", cluster.Name, err)
 		}
 
-		// Create tokens for all of the agents to use for anti-entropy.
-		//
-		// NOTE: this will cause the servers to roll to pick up the change to
-		// the acl{tokens{agent=XXX}}} section.
-		if err := s.createAgentTokens(cluster); err != nil {
-			return fmt.Errorf("createAgentTokens[%s]: %w", cluster.Name, err)
+		if node.Images.GreaterThanVersion(topology.MinVersionAgentTokenPartition) {
+			// Create tokens for all of the agents to use for anti-entropy.
+			//
+			// NOTE: this will cause the servers to roll to pick up the change to
+			// the acl{tokens{agent=XXX}}} section.
+			if err := s.createAgentTokens(cluster); err != nil {
+				return fmt.Errorf("createAgentTokens[%s]: %w", cluster.Name, err)
+			}
+		} else {
+			// Assign agent join policy to the anonymous token
+			if err := s.assignAgentJoinPolicyToAnonymousToken(cluster); err != nil {
+				return fmt.Errorf("assignAgentJoinPolicyToAnonymousToken[%s]: %w", cluster.Name, err)
+			}
 		}
 	}
 
@@ -487,7 +494,8 @@ func (s *Sprawl) waitForLocalWrites(cluster *topology.Cluster, token string) {
 		break
 	}
 
-	if cluster.Enterprise {
+	serverNodes := cluster.ServerNodes()
+	if cluster.Enterprise && serverNodes[0].Images.GreaterThanVersion(topology.MinVersionAgentTokenPartition) {
 		start = time.Now()
 		for attempts := 0; ; attempts++ {
 			if err := tryAP(); err != nil {
@@ -543,7 +551,7 @@ func (s *Sprawl) waitForClientAntiEntropyOnce(cluster *topology.Cluster) error {
 			nid := node.CatalogID()
 
 			got, ok := current[nid]
-			if ok && len(got.TaggedAddresses) > 0 {
+			if ok && (len(got.TaggedAddresses) > 0 || got.Address != "") {
 				// this is a field that is not updated just due to serf reconcile
 				continue
 			}
diff --git a/testing/deployer/sprawl/internal/tfgen/agent.go b/testing/deployer/sprawl/internal/tfgen/agent.go
index 8e62920930..db68a60b61 100644
--- a/testing/deployer/sprawl/internal/tfgen/agent.go
+++ b/testing/deployer/sprawl/internal/tfgen/agent.go
@@ -60,10 +60,12 @@ func (g *Generator) generateAgentHCL(node *topology.Node, enableV2, enableV2Tena
 	b.add("retry_interval", "1s")
 	// }
 
-	if node.IsServer() {
-		b.addBlock("peering", func() {
-			b.add("enabled", true)
-		})
+	if node.Images.GreaterThanVersion(topology.MinVersionPeering) {
+		if node.IsServer() {
+			b.addBlock("peering", func() {
+				b.add("enabled", true)
+			})
+		}
 	}
 
 	b.addBlock("ui_config", func() {
@@ -85,41 +87,45 @@ func (g *Generator) generateAgentHCL(node *topology.Node, enableV2, enableV2Tena
 			certKey  = root + "/" + node.TLSCertPrefix + "-key.pem"
 		)
 
-		b.addBlock("tls", func() {
-			b.addBlock("internal_rpc", func() {
-				b.add("ca_file", caFile)
-				b.add("cert_file", certFile)
-				b.add("key_file", certKey)
-				b.add("verify_incoming", true)
-				b.add("verify_server_hostname", true)
-				b.add("verify_outgoing", true)
-			})
-			// if cfg.EncryptionTLSAPI {
-			// 	b.addBlock("https", func() {
-			// 		b.add("ca_file", caFile)
-			// 		b.add("cert_file", certFile)
-			// 		b.add("key_file", certKey)
-			// 		// b.add("verify_incoming", true)
-			// 	})
-			// }
-			if node.IsServer() {
-				b.addBlock("grpc", func() {
+		if node.Images.GreaterThanVersion(topology.MinVersionTLS) {
+			b.addBlock("tls", func() {
+				b.addBlock("internal_rpc", func() {
 					b.add("ca_file", caFile)
 					b.add("cert_file", certFile)
 					b.add("key_file", certKey)
-					// b.add("verify_incoming", true)
+					b.add("verify_incoming", true)
+					b.add("verify_server_hostname", true)
+					b.add("verify_outgoing", true)
 				})
-			}
-		})
+				// if cfg.EncryptionTLSAPI {
+				// 	b.addBlock("https", func() {
+				// 		b.add("ca_file", caFile)
+				// 		b.add("cert_file", certFile)
+				// 		b.add("key_file", certKey)
+				// 		// b.add("verify_incoming", true)
+				// 	})
+				// }
+				if node.IsServer() {
+					b.addBlock("grpc", func() {
+						b.add("ca_file", caFile)
+						b.add("cert_file", certFile)
+						b.add("key_file", certKey)
+						// b.add("verify_incoming", true)
+					})
+				}
+			})
+		}
 	}
 
 	b.addBlock("ports", func() {
-		if node.IsServer() {
-			b.add("grpc_tls", 8503)
-			b.add("grpc", -1)
-		} else {
-			b.add("grpc", 8502)
-			b.add("grpc_tls", -1)
+		if node.Images.GreaterThanVersion(topology.MinVersionPeering) {
+			if node.IsServer() {
+				b.add("grpc_tls", 8503)
+				b.add("grpc", -1)
+			} else {
+				b.add("grpc", 8502)
+				b.add("grpc_tls", -1)
+			}
 		}
 		b.add("http", 8500)
 		b.add("dns", 8600)
@@ -132,13 +138,22 @@ func (g *Generator) generateAgentHCL(node *topology.Node, enableV2, enableV2Tena
 		b.add("default_policy", "deny")
 		b.add("down_policy", "extend-cache")
 		b.add("enable_token_persistence", true)
-		b.addBlock("tokens", func() {
-			if node.IsServer() {
-				b.add("initial_management", g.sec.ReadGeneric(node.Cluster, secrets.BootstrapToken))
-			}
-			b.add("agent_recovery", g.sec.ReadGeneric(node.Cluster, secrets.AgentRecovery))
-			b.add("agent", g.sec.ReadAgentToken(node.Cluster, node.ID()))
-		})
+
+		if node.Images.GreaterThanVersion(topology.MinVersionAgentTokenPartition) {
+			b.addBlock("tokens", func() {
+				if node.IsServer() {
+					b.add("initial_management", g.sec.ReadGeneric(node.Cluster, secrets.BootstrapToken))
+				}
+				b.add("agent_recovery", g.sec.ReadGeneric(node.Cluster, secrets.AgentRecovery))
+				b.add("agent", g.sec.ReadAgentToken(node.Cluster, node.ID()))
+			})
+		} else {
+			b.addBlock("tokens", func() {
+				if node.IsServer() {
+					b.add("master", g.sec.ReadGeneric(node.Cluster, secrets.BootstrapToken))
+				}
+			})
+		}
 	})
 
 	if node.IsServer() {
@@ -195,7 +210,7 @@ func (g *Generator) generateAgentHCL(node *topology.Node, enableV2, enableV2Tena
 			})
 		}
 	} else {
-		if cluster.Enterprise {
+		if cluster.Enterprise && node.Images.GreaterThanVersion(topology.MinVersionAgentTokenPartition) {
 			b.add("partition", node.Partition)
 		}
 	}
diff --git a/testing/deployer/sprawl/sprawl.go b/testing/deployer/sprawl/sprawl.go
index b622f986af..bb57e4b8c8 100644
--- a/testing/deployer/sprawl/sprawl.go
+++ b/testing/deployer/sprawl/sprawl.go
@@ -299,6 +299,18 @@ func (s *Sprawl) standardUpgrade(cluster *topology.Cluster,
 			return fmt.Errorf("error upgrading node %s: %w", node.Name, err)
 		}
 	}
+
+	// upgrade client agents one at a time
+	for _, node := range cluster.Nodes {
+		if node.Kind != topology.NodeKindClient {
+			s.logger.Info("Skip non-client node", "node", node.Name)
+			continue
+		}
+		if err := upgradeFn(node.ID()); err != nil {
+			return fmt.Errorf("error upgrading node %s: %w", node.Name, err)
+		}
+	}
+
 	return nil
 }
 
diff --git a/testing/deployer/topology/images.go b/testing/deployer/topology/images.go
index 47a8956b4a..5692e6bd78 100644
--- a/testing/deployer/topology/images.go
+++ b/testing/deployer/topology/images.go
@@ -5,6 +5,14 @@ package topology
 
 import (
 	"strings"
+
+	goversion "github.com/hashicorp/go-version"
+)
+
+var (
+	MinVersionAgentTokenPartition = goversion.Must(goversion.NewVersion("v1.11.0"))
+	MinVersionPeering             = goversion.Must(goversion.NewVersion("v1.13.0"))
+	MinVersionTLS                 = goversion.Must(goversion.NewVersion("v1.12.0"))
 )
 
 type Images struct {
@@ -13,6 +21,10 @@ type Images struct {
 	Consul string `json:",omitempty"`
 	// ConsulCE sets the CE image
 	ConsulCE string `json:",omitempty"`
+	// consulVersion is the version part of Consul image,
+	// e.g., if Consul image is hashicorp/consul-enterprise:1.15.0-ent,
+	// consulVersion is 1.15.0-ent
+	consulVersion string
 	// ConsulEnterprise sets the ent image
 	ConsulEnterprise string `json:",omitempty"`
 	Envoy            string
@@ -93,9 +105,21 @@ func (i Images) ChooseConsul(enterprise bool) Images {
 	}
 	i.ConsulEnterprise = ""
 	i.ConsulCE = ""
+
+	// extract the version part of Consul
+	i.consulVersion = i.Consul[strings.Index(i.Consul, ":")+1:]
 	return i
 }
 
+// GreaterThanVersion compares the image version to a specified version
+func (i Images) GreaterThanVersion(version *goversion.Version) bool {
+	if i.consulVersion == "local" {
+		return true
+	}
+	iVer := goversion.Must(goversion.NewVersion(i.consulVersion))
+	return iVer.GreaterThanOrEqual(version)
+}
+
 func (i Images) OverrideWith(i2 Images) Images {
 	if i2.Consul != "" {
 		i.Consul = i2.Consul