diff --git a/agent/config/builder.go b/agent/config/builder.go
index 2ead17543e86..513d5931c7b5 100644
--- a/agent/config/builder.go
+++ b/agent/config/builder.go
@@ -2527,18 +2527,23 @@ func validateAutoConfigAuthorizer(rt RuntimeConfig) error {
 	return nil
 }
 
-func (b *builder) cloudConfigVal(v *CloudConfigRaw) (val hcpconfig.CloudConfig) {
+func (b *builder) cloudConfigVal(v *CloudConfigRaw) hcpconfig.CloudConfig {
+	val := hcpconfig.CloudConfig{
+		ResourceID: os.Getenv("HCP_RESOURCE_ID"),
+	}
 	if v == nil {
 		return val
 	}
 
-	val.ResourceID = stringVal(v.ResourceID)
 	val.ClientID = stringVal(v.ClientID)
 	val.ClientSecret = stringVal(v.ClientSecret)
 	val.AuthURL = stringVal(v.AuthURL)
 	val.Hostname = stringVal(v.Hostname)
 	val.ScadaAddress = stringVal(v.ScadaAddress)
 
+	if resourceID := stringVal(v.ResourceID); resourceID != "" {
+		val.ResourceID = resourceID
+	}
 	return val
 }
 
diff --git a/agent/config/runtime.go b/agent/config/runtime.go
index 790eec425a94..a8d6c62ebde9 100644
--- a/agent/config/runtime.go
+++ b/agent/config/runtime.go
@@ -1749,6 +1749,9 @@ func (c *RuntimeConfig) Sanitized() map[string]interface{} {
 
 // IsCloudEnabled returns true if a cloud.resource_id is set and the server mode is enabled
 func (c *RuntimeConfig) IsCloudEnabled() bool {
+	if c == nil {
+		return false
+	}
 	return c.ServerMode && c.Cloud.ResourceID != ""
 }
 
diff --git a/agent/config/runtime_test.go b/agent/config/runtime_test.go
index cd1a83cbb961..8b8ee4f86a10 100644
--- a/agent/config/runtime_test.go
+++ b/agent/config/runtime_test.go
@@ -2301,6 +2301,74 @@ func TestLoad_IntegrationWithFlags(t *testing.T) {
 			rt.HTTPUseCache = false
 		},
 	})
+	run(t, testCase{
+		desc: "cloud resource id from env",
+		args: []string{
+			`-server`,
+			`-data-dir=` + dataDir,
+		},
+		setup: func() {
+			os.Setenv("HCP_RESOURCE_ID", "env-id")
+			t.Cleanup(func() {
+				os.Unsetenv("HCP_RESOURCE_ID")
+			})
+		},
+		expected: func(rt *RuntimeConfig) {
+			rt.DataDir = dataDir
+			rt.Cloud = hcpconfig.CloudConfig{
+				// ID is only populated from env if not populated from other sources.
+				ResourceID: "env-id",
+			}
+
+			// server things
+			rt.ServerMode = true
+			rt.TLS.ServerMode = true
+			rt.LeaveOnTerm = false
+			rt.SkipLeaveOnInt = true
+			rt.RPCConfig.EnableStreaming = true
+			rt.GRPCTLSPort = 8503
+			rt.GRPCTLSAddrs = []net.Addr{defaultGrpcTlsAddr}
+		},
+	})
+	run(t, testCase{
+		desc: "cloud resource id from file",
+		args: []string{
+			`-server`,
+			`-data-dir=` + dataDir,
+		},
+		setup: func() {
+			os.Setenv("HCP_RESOURCE_ID", "env-id")
+			t.Cleanup(func() {
+				os.Unsetenv("HCP_RESOURCE_ID")
+			})
+		},
+		json: []string{`{
+			  "cloud": {
+              	"resource_id": "file-id" 
+              }
+			}`},
+		hcl: []string{`
+			  cloud = {
+	            resource_id = "file-id" 
+			  }
+			`},
+		expected: func(rt *RuntimeConfig) {
+			rt.DataDir = dataDir
+			rt.Cloud = hcpconfig.CloudConfig{
+				// ID is only populated from env if not populated from other sources.
+				ResourceID: "file-id",
+			}
+
+			// server things
+			rt.ServerMode = true
+			rt.TLS.ServerMode = true
+			rt.LeaveOnTerm = false
+			rt.SkipLeaveOnInt = true
+			rt.RPCConfig.EnableStreaming = true
+			rt.GRPCTLSPort = 8503
+			rt.GRPCTLSAddrs = []net.Addr{defaultGrpcTlsAddr}
+		},
+	})
 	run(t, testCase{
 		desc: "sidecar_service can't have ID",
 		args: []string{
diff --git a/agent/config/testdata/TestRuntimeConfig_Sanitize.golden b/agent/config/testdata/TestRuntimeConfig_Sanitize.golden
index 054a284aeb85..f2bd8f2f3ca5 100644
--- a/agent/config/testdata/TestRuntimeConfig_Sanitize.golden
+++ b/agent/config/testdata/TestRuntimeConfig_Sanitize.golden
@@ -131,8 +131,10 @@
         "ClientID": "id",
         "ClientSecret": "hidden",
         "Hostname": "",
+        "ManagementToken": "hidden",
         "ResourceID": "cluster1",
-        "ScadaAddress": ""
+        "ScadaAddress": "",
+        "TLSConfig": null
     },
     "ConfigEntryBootstrap": [],
     "ConnectCAConfig": {},
diff --git a/agent/hcp/bootstrap/bootstrap.go b/agent/hcp/bootstrap/bootstrap.go
index 55e1231f5cf4..e6a1ae120a6b 100644
--- a/agent/hcp/bootstrap/bootstrap.go
+++ b/agent/hcp/bootstrap/bootstrap.go
@@ -10,7 +10,10 @@ package bootstrap
 import (
 	"bufio"
 	"context"
+	"crypto/tls"
+	"crypto/x509"
 	"encoding/json"
+	"encoding/pem"
 	"errors"
 	"fmt"
 	"os"
@@ -19,17 +22,22 @@ import (
 	"time"
 
 	"github.com/hashicorp/consul/agent/config"
+	"github.com/hashicorp/consul/agent/connect"
 	"github.com/hashicorp/consul/agent/hcp"
 	"github.com/hashicorp/consul/lib"
 	"github.com/hashicorp/consul/lib/retry"
+	"github.com/hashicorp/go-uuid"
 )
 
 const (
-	caFileName     = "server-tls-cas.pem"
-	certFileName   = "server-tls-cert.pem"
-	keyFileName    = "server-tls-key.pem"
-	configFileName = "server-config.json"
-	subDir         = "hcp-config"
+	subDir = "hcp-config"
+
+	caFileName      = "server-tls-cas.pem"
+	certFileName    = "server-tls-cert.pem"
+	configFileName  = "server-config.json"
+	keyFileName     = "server-tls-key.pem"
+	tokenFileName   = "hcp-management-token"
+	successFileName = "successful-bootstrap"
 )
 
 type ConfigLoader func(source config.Source) (config.LoadResult, error)
@@ -46,44 +54,80 @@ type UI interface {
 	Error(string)
 }
 
-// MaybeBootstrap will use the passed ConfigLoader to read the existing
-// configuration, and if required attempt to bootstrap from HCP. It will retry
-// until successful or a terminal error condition is found (e.g. permission
-// denied). It must be passed a (CLI) UI implementation so it can deliver progress
-// updates to the user, for example if it is waiting to retry for a long period.
-func MaybeBootstrap(ctx context.Context, loader ConfigLoader, ui UI) (bool, ConfigLoader, error) {
-	loader = wrapConfigLoader(loader)
-	res, err := loader(nil)
-	if err != nil {
-		return false, nil, err
-	}
-
-	// Check to see if this is a server and HCP is configured
-
-	if !res.RuntimeConfig.IsCloudEnabled() {
-		// Not a server, let agent continue unmodified
-		return false, loader, nil
-	}
+// RawBootstrapConfig contains the Consul config as a raw JSON string and the management token
+// which either was retrieved from persisted files or from the bootstrap endpoint
+type RawBootstrapConfig struct {
+	ConfigJSON      string
+	ManagementToken string
+}
 
-	ui.Output("Bootstrapping configuration from HCP")
+// LoadConfig will attempt to load previously-fetched config from disk and fall back to
+// fetch from HCP servers if the local data is incomplete.
+// It must be passed a (CLI) UI implementation so it can deliver progress
+// updates to the user, for example if it is waiting to retry for a long period.
+func LoadConfig(ctx context.Context, client hcp.Client, dataDir string, loader ConfigLoader, ui UI) (ConfigLoader, error) {
+	ui.Output("Loading configuration from HCP")
 
 	// See if we have existing config on disk
-	cfgJSON, ok := loadPersistedBootstrapConfig(res.RuntimeConfig, ui)
-
+	//
+	// OPTIMIZE: We could probably be more intelligent about config loading.
+	// The currently implemented approach is:
+	// 1. Attempt to load data from disk
+	// 2. If that fails or the data is incomplete, block indefinitely fetching remote config.
+	//
+	// What if instead we had the following flow:
+	// 1. Attempt to fetch config from HCP.
+	// 2. If that fails, fall back to data on disk from last fetch.
+	// 3. If that fails, go into blocking loop to fetch remote config.
+	//
+	// This should allow us to more gracefully transition cases like when
+	// an existing cluster is linked, but then wants to receive TLS materials
+	// at a later time. Currently, if we observe the existing-cluster marker we
+	// don't attempt to fetch any additional configuration from HCP.
+
+	cfg, ok := loadPersistedBootstrapConfig(dataDir, ui)
 	if !ok {
-		// Fetch from HCP
-		ui.Info("Fetching configuration from HCP")
-		cfgJSON, err = doHCPBootstrap(ctx, res.RuntimeConfig, ui)
+		ui.Info("Fetching configuration from HCP servers")
+
+		var err error
+		cfg, err = fetchBootstrapConfig(ctx, client, dataDir, ui)
 		if err != nil {
-			return false, nil, fmt.Errorf("failed to bootstrap from HCP: %w", err)
+			return nil, fmt.Errorf("failed to bootstrap from HCP: %w", err)
 		}
 		ui.Info("Configuration fetched from HCP and saved on local disk")
+
 	} else {
-		ui.Info("Loaded configuration from local disk")
+		ui.Info("Loaded HCP configuration from local disk")
+
 	}
 
 	// Create a new loader func to return
-	newLoader := func(source config.Source) (config.LoadResult, error) {
+	newLoader := bootstrapConfigLoader(loader, cfg)
+	return newLoader, nil
+}
+
+// bootstrapConfigLoader is a ConfigLoader for passing bootstrap JSON config received from HCP
+// to the config.builder. ConfigLoaders are functions used to build an agent's RuntimeConfig
+// from various sources like files and flags. This config is contained in the config.LoadResult.
+//
+// The flow to include bootstrap config from HCP as a loader's data source is as follows:
+//
+//  1. A base ConfigLoader function (baseLoader) is created on agent start, and it sets the input
+//     source argument as the DefaultConfig.
+//
+//  2. When a server agent can be configured by HCP that baseLoader is wrapped in this bootstrapConfigLoader.
+//
+//  3. The bootstrapConfigLoader calls that base loader with the bootstrap JSON config as the
+//     default source. This data will be merged with other valid sources in the config.builder.
+//
+//  4. The result of the call to baseLoader() below contains the resulting RuntimeConfig, and we do some
+//     additional modifications to attach data that doesn't get populated during the build in the config pkg.
+//
+// Note that since the ConfigJSON is stored as the baseLoader's DefaultConfig, its data is the first
+// to be merged by the config.builder and could be overwritten by user-provided values in config files or
+// CLI flags. However, values set to RuntimeConfig after the baseLoader call are final.
+func bootstrapConfigLoader(baseLoader ConfigLoader, cfg *RawBootstrapConfig) ConfigLoader {
+	return func(source config.Source) (config.LoadResult, error) {
 		// Don't allow any further attempts to provide a DefaultSource. This should
 		// only ever be needed later in client agent AutoConfig code but that should
 		// be mutually exclusive from this bootstrapping mechanism since this is
@@ -94,34 +138,50 @@ func MaybeBootstrap(ctx context.Context, loader ConfigLoader, ui UI) (bool, Conf
 			return config.LoadResult{},
 				fmt.Errorf("non-nil config source provided to a loader after HCP bootstrap already provided a DefaultSource")
 		}
+
 		// Otherwise, just call to the loader we were passed with our own additional
 		// JSON as the source.
-		s := config.FileSource{
+		//
+		// OPTIMIZE: We could check/log whether any fields set by the remote config were overwritten by a user-provided flag.
+		res, err := baseLoader(config.FileSource{
 			Name:   "HCP Bootstrap",
 			Format: "json",
-			Data:   cfgJSON,
+			Data:   cfg.ConfigJSON,
+		})
+		if err != nil {
+			return res, fmt.Errorf("failed to load HCP Bootstrap config: %w", err)
 		}
-		return loader(s)
-	}
 
-	return true, newLoader, nil
+		finalizeRuntimeConfig(res.RuntimeConfig, cfg)
+		return res, nil
+	}
 }
 
-func wrapConfigLoader(loader ConfigLoader) ConfigLoader {
-	return func(source config.Source) (config.LoadResult, error) {
-		res, err := loader(source)
-		if err != nil {
-			return res, err
-		}
+const (
+	accessControlHeaderName  = "Access-Control-Expose-Headers"
+	accessControlHeaderValue = "x-consul-default-acl-policy"
+)
 
-		if res.RuntimeConfig.Cloud.ResourceID == "" {
-			res.RuntimeConfig.Cloud.ResourceID = os.Getenv("HCP_RESOURCE_ID")
-		}
-		return res, nil
+// finalizeRuntimeConfig will set additional HCP-specific values that are not
+// handled by the config.builder.
+func finalizeRuntimeConfig(rc *config.RuntimeConfig, cfg *RawBootstrapConfig) {
+	rc.Cloud.ManagementToken = cfg.ManagementToken
+
+	// HTTP response headers are modified for the HCP UI to work.
+	if rc.HTTPResponseHeaders == nil {
+		rc.HTTPResponseHeaders = make(map[string]string)
+	}
+	prevValue, ok := rc.HTTPResponseHeaders[accessControlHeaderName]
+	if !ok {
+		rc.HTTPResponseHeaders[accessControlHeaderName] = accessControlHeaderValue
+	} else {
+		rc.HTTPResponseHeaders[accessControlHeaderName] = prevValue + "," + accessControlHeaderValue
 	}
 }
 
-func doHCPBootstrap(ctx context.Context, rc *config.RuntimeConfig, ui UI) (string, error) {
+// fetchBootstrapConfig will fetch boostrap configuration from remote servers and persist it to disk.
+// It will retry until successful or a terminal error condition is found (e.g. permission denied).
+func fetchBootstrapConfig(ctx context.Context, client hcp.Client, dataDir string, ui UI) (*RawBootstrapConfig, error) {
 	w := retry.Waiter{
 		MinWait: 1 * time.Second,
 		MaxWait: 5 * time.Minute,
@@ -129,12 +189,6 @@ func doHCPBootstrap(ctx context.Context, rc *config.RuntimeConfig, ui UI) (strin
 	}
 
 	var bsCfg *hcp.BootstrapConfig
-
-	client, err := hcp.NewClient(rc.Cloud)
-	if err != nil {
-		return "", err
-	}
-
 	for {
 		// Note we don't want to shadow `ctx` here since we need that for the Wait
 		// below.
@@ -143,10 +197,10 @@ func doHCPBootstrap(ctx context.Context, rc *config.RuntimeConfig, ui UI) (strin
 
 		resp, err := client.FetchBootstrap(reqCtx)
 		if err != nil {
-			ui.Error(fmt.Sprintf("failed to fetch bootstrap config from HCP, will retry in %s: %s",
+			ui.Error(fmt.Sprintf("Error: failed to fetch bootstrap config from HCP, will retry in %s: %s",
 				w.NextWait().Round(time.Second), err))
 			if err := w.Wait(ctx); err != nil {
-				return "", err
+				return nil, err
 			}
 			// Finished waiting, restart loop
 			continue
@@ -155,9 +209,24 @@ func doHCPBootstrap(ctx context.Context, rc *config.RuntimeConfig, ui UI) (strin
 		break
 	}
 
-	dataDir := rc.DataDir
-	shouldPersist := true
-	if dataDir == "" {
+	devMode := dataDir == ""
+
+	cfgJSON, err := persistAndProcessConfig(dataDir, devMode, bsCfg)
+	if err != nil {
+		return nil, fmt.Errorf("failed to persist config for existing cluster: %w", err)
+	}
+
+	return &RawBootstrapConfig{
+		ConfigJSON:      cfgJSON,
+		ManagementToken: bsCfg.ManagementToken,
+	}, nil
+}
+
+// persistAndProcessConfig is called when we receive data from CCM.
+// We validate and persist everything that was received, then also update
+// the JSON config as needed.
+func persistAndProcessConfig(dataDir string, devMode bool, bsCfg *hcp.BootstrapConfig) (string, error) {
+	if devMode {
 		// Agent in dev mode, we still need somewhere to persist the certs
 		// temporarily though to be able to start up at all since we don't support
 		// inline certs right now. Use temp dir
@@ -166,41 +235,92 @@ func doHCPBootstrap(ctx context.Context, rc *config.RuntimeConfig, ui UI) (strin
 			return "", fmt.Errorf("failed to create temp dir for certificates: %w", err)
 		}
 		dataDir = tmp
-		shouldPersist = false
 	}
 
-	// Persist the TLS cert files from the response since we need to refer to them
-	// as disk files either way.
-	if err := persistTLSCerts(dataDir, bsCfg); err != nil {
-		return "", fmt.Errorf("failed to persist TLS certificates to dir %q: %w", dataDir, err)
+	// Create subdir if it's not already there.
+	dir := filepath.Join(dataDir, subDir)
+	if err := lib.EnsurePath(dir, true); err != nil {
+		return "", fmt.Errorf("failed to ensure directory %q: %w", dir, err)
 	}
-	// Update the config JSON to include those TLS cert files
-	cfgJSON, err := injectTLSCerts(dataDir, bsCfg.ConsulConfig)
-	if err != nil {
-		return "", fmt.Errorf("failed to inject TLS Certs into bootstrap config: %w", err)
+
+	// Parse just to a map for now as we only have to inject to a specific place
+	// and parsing whole Config struct is complicated...
+	var cfg map[string]any
+
+	if err := json.Unmarshal([]byte(bsCfg.ConsulConfig), &cfg); err != nil {
+		return "", fmt.Errorf("failed to unmarshal bootstrap config: %w", err)
+	}
+
+	// Avoid ever setting an initial_management token from HCP now that we can
+	// separately bootstrap an HCP management token with a distinct accessor ID.
+	//
+	// CCM will continue to return an initial_management token because previous versions of Consul
+	// cannot bootstrap an HCP management token distinct from the initial management token.
+	// This block can be deleted once CCM supports tailoring bootstrap config responses
+	// based on the version of Consul that requested it.
+	acls, aclsOK := cfg["acl"].(map[string]any)
+	if aclsOK {
+		tokens, tokensOK := acls["tokens"].(map[string]interface{})
+		if tokensOK {
+			delete(tokens, "initial_management")
+		}
 	}
 
-	// Persist the final config we need to add for restarts. Assuming this wasn't
-	// a tmp dir to start with.
-	if shouldPersist {
-		if err := persistBootstrapConfig(dataDir, cfgJSON); err != nil {
-			return "", fmt.Errorf("failed to persist bootstrap config to dir %q: %w", dataDir, err)
+	var cfgJSON string
+	if bsCfg.TLSCert != "" {
+		if err := validateTLSCerts(bsCfg.TLSCert, bsCfg.TLSCertKey, bsCfg.TLSCAs); err != nil {
+			return "", fmt.Errorf("invalid certificates: %w", err)
+		}
+
+		// Persist the TLS cert files from the response since we need to refer to them
+		// as disk files either way.
+		if err := persistTLSCerts(dir, bsCfg.TLSCert, bsCfg.TLSCertKey, bsCfg.TLSCAs); err != nil {
+			return "", fmt.Errorf("failed to persist TLS certificates to dir %q: %w", dataDir, err)
+		}
+
+		// Store paths to the persisted TLS cert files.
+		cfg["ca_file"] = filepath.Join(dir, caFileName)
+		cfg["cert_file"] = filepath.Join(dir, certFileName)
+		cfg["key_file"] = filepath.Join(dir, keyFileName)
+
+		// Convert the bootstrap config map back into a string
+		cfgJSONBytes, err := json.Marshal(cfg)
+		if err != nil {
+			return "", err
 		}
+		cfgJSON = string(cfgJSONBytes)
 	}
 
+	if !devMode {
+		// Persist the final config we need to add so that it is available locally after a restart.
+		// Assuming the configured data dir wasn't a tmp dir to start with.
+		if err := persistBootstrapConfig(dir, cfgJSON); err != nil {
+			return "", fmt.Errorf("failed to persist bootstrap config: %w", err)
+		}
+
+		if err := validateManagementToken(bsCfg.ManagementToken); err != nil {
+			return "", fmt.Errorf("invalid management token: %w", err)
+		}
+		if err := persistManagementToken(dir, bsCfg.ManagementToken); err != nil {
+			return "", fmt.Errorf("failed to persist HCP management token: %w", err)
+		}
+
+		if err := persistSucessMarker(dir); err != nil {
+			return "", fmt.Errorf("failed to persist success marker: %w", err)
+		}
+	}
 	return cfgJSON, nil
 }
 
-func persistTLSCerts(dataDir string, bsCfg *hcp.BootstrapConfig) error {
-	dir := filepath.Join(dataDir, subDir)
+func persistSucessMarker(dir string) error {
+	name := filepath.Join(dir, successFileName)
+	return os.WriteFile(name, []byte(""), 0600)
 
-	if bsCfg.TLSCert == "" || bsCfg.TLSCertKey == "" {
-		return fmt.Errorf("unexpected bootstrap response from HCP: missing TLS information")
-	}
+}
 
-	// Create a subdir if it's not already there
-	if err := lib.EnsurePath(dir, true); err != nil {
-		return err
+func persistTLSCerts(dir string, serverCert, serverKey string, caCerts []string) error {
+	if serverCert == "" || serverKey == "" {
+		return fmt.Errorf("unexpected bootstrap response from HCP: missing TLS information")
 	}
 
 	// Write out CA cert(s). We write them all to one file because Go's x509
@@ -211,7 +331,7 @@ func persistTLSCerts(dataDir string, bsCfg *hcp.BootstrapConfig) error {
 		return err
 	}
 	bf := bufio.NewWriter(f)
-	for _, caPEM := range bsCfg.TLSCAs {
+	for _, caPEM := range caCerts {
 		bf.WriteString(caPEM + "\n")
 	}
 	if err := bf.Flush(); err != nil {
@@ -221,87 +341,243 @@ func persistTLSCerts(dataDir string, bsCfg *hcp.BootstrapConfig) error {
 		return err
 	}
 
-	if err := os.WriteFile(filepath.Join(dir, certFileName), []byte(bsCfg.TLSCert), 0600); err != nil {
+	if err := os.WriteFile(filepath.Join(dir, certFileName), []byte(serverCert), 0600); err != nil {
 		return err
 	}
 
-	if err := os.WriteFile(filepath.Join(dir, keyFileName), []byte(bsCfg.TLSCertKey), 0600); err != nil {
+	if err := os.WriteFile(filepath.Join(dir, keyFileName), []byte(serverKey), 0600); err != nil {
 		return err
 	}
 
 	return nil
 }
 
-func injectTLSCerts(dataDir string, bootstrapJSON string) (string, error) {
-	// Parse just to a map for now as we only have to inject to a specific place
-	// and parsing whole Config struct is complicated...
-	var cfg map[string]interface{}
-
-	if err := json.Unmarshal([]byte(bootstrapJSON), &cfg); err != nil {
-		return "", err
+// Basic validation to ensure a UUID was loaded.
+func validateManagementToken(token string) error {
+	if token == "" {
+		return errors.New("missing HCP management token")
 	}
 
-	// Inject TLS cert files
-	cfg["ca_file"] = filepath.Join(dataDir, subDir, caFileName)
-	cfg["cert_file"] = filepath.Join(dataDir, subDir, certFileName)
-	cfg["key_file"] = filepath.Join(dataDir, subDir, keyFileName)
-
-	jsonBs, err := json.Marshal(cfg)
-	if err != nil {
-		return "", err
+	if _, err := uuid.ParseUUID(token); err != nil {
+		return errors.New("management token is not a valid UUID")
 	}
+	return nil
+}
 
-	return string(jsonBs), nil
+func persistManagementToken(dir, token string) error {
+	name := filepath.Join(dir, tokenFileName)
+	return os.WriteFile(name, []byte(token), 0600)
 }
 
-func persistBootstrapConfig(dataDir, cfgJSON string) error {
+func persistBootstrapConfig(dir, cfgJSON string) error {
 	// Persist the important bits we got from bootstrapping. The TLS certs are
 	// already persisted, just need to persist the config we are going to add.
-	name := filepath.Join(dataDir, subDir, configFileName)
+	name := filepath.Join(dir, configFileName)
 	return os.WriteFile(name, []byte(cfgJSON), 0600)
 }
 
-func loadPersistedBootstrapConfig(rc *config.RuntimeConfig, ui UI) (string, bool) {
-	// Check if the files all exist
+func loadPersistedBootstrapConfig(dataDir string, ui UI) (*RawBootstrapConfig, bool) {
+	if dataDir == "" {
+		// There's no files to load when in dev mode.
+		return nil, false
+	}
+
+	dir := filepath.Join(dataDir, subDir)
+
+	_, err := os.Stat(filepath.Join(dir, successFileName))
+	if os.IsNotExist(err) {
+		// Haven't bootstrapped from HCP.
+		return nil, false
+	}
+	if err != nil {
+		ui.Warn("failed to check for config on disk, re-fetching from HCP: " + err.Error())
+		return nil, false
+	}
+
+	if err := checkCerts(dir); err != nil {
+		ui.Warn("failed to validate certs on disk, re-fetching from HCP: " + err.Error())
+		return nil, false
+	}
+
+	configJSON, err := loadBootstrapConfigJSON(dataDir)
+	if err != nil {
+		ui.Warn("failed to load bootstrap config from disk, re-fetching from HCP: " + err.Error())
+		return nil, false
+	}
+
+	mgmtToken, err := loadManagementToken(dir)
+	if err != nil {
+		ui.Warn("failed to load HCP management token from disk, re-fetching from HCP: " + err.Error())
+		return nil, false
+	}
+
+	return &RawBootstrapConfig{
+		ConfigJSON:      configJSON,
+		ManagementToken: mgmtToken,
+	}, true
+}
+
+func loadBootstrapConfigJSON(dataDir string) (string, error) {
+	filename := filepath.Join(dataDir, subDir, configFileName)
+
+	_, err := os.Stat(filename)
+	if os.IsNotExist(err) {
+		return "", nil
+	}
+	if err != nil {
+		return "", fmt.Errorf("failed to check for bootstrap config: %w", err)
+	}
+
+	// Attempt to load persisted config to check for errors and basic validity.
+	// Errors here will raise issues like referencing unsupported config fields.
+	_, err = config.Load(config.LoadOpts{
+		ConfigFiles: []string{filename},
+		HCL: []string{
+			"server = true",
+			`bind_addr = "127.0.0.1"`,
+			fmt.Sprintf("data_dir = %q", dataDir),
+		},
+		ConfigFormat: "json",
+	})
+	if err != nil {
+		return "", fmt.Errorf("failed to parse local bootstrap config: %w", err)
+	}
+
+	jsonBs, err := os.ReadFile(filename)
+	if err != nil {
+		return "", fmt.Errorf(fmt.Sprintf("failed to read local bootstrap config file: %s", err))
+	}
+	return strings.TrimSpace(string(jsonBs)), nil
+}
+
+func loadManagementToken(dir string) (string, error) {
+	name := filepath.Join(dir, tokenFileName)
+	bytes, err := os.ReadFile(name)
+	if os.IsNotExist(err) {
+		return "", errors.New("configuration files on disk are incomplete, missing: " + name)
+	}
+	if err != nil {
+		return "", fmt.Errorf("failed to read: %w", err)
+	}
+
+	token := string(bytes)
+	if err := validateManagementToken(token); err != nil {
+		return "", fmt.Errorf("invalid management token: %w", err)
+	}
+
+	return token, nil
+}
+
+func checkCerts(dir string) error {
 	files := []string{
-		filepath.Join(rc.DataDir, subDir, configFileName),
-		filepath.Join(rc.DataDir, subDir, caFileName),
-		filepath.Join(rc.DataDir, subDir, certFileName),
-		filepath.Join(rc.DataDir, subDir, keyFileName),
-	}
-	hasSome := false
-	for _, name := range files {
-		if _, err := os.Stat(name); errors.Is(err, os.ErrNotExist) {
-			// At least one required file doesn't exist, failed loading. This is not
-			// an error though
-			if hasSome {
-				ui.Warn("ignoring incomplete local bootstrap config files")
-			}
-			return "", false
+		filepath.Join(dir, caFileName),
+		filepath.Join(dir, certFileName),
+		filepath.Join(dir, keyFileName),
+	}
+
+	missing := make([]string, 0)
+	for _, file := range files {
+		_, err := os.Stat(file)
+		if os.IsNotExist(err) {
+			missing = append(missing, file)
+			continue
+		}
+		if err != nil {
+			return err
 		}
-		hasSome = true
 	}
 
-	name := filepath.Join(rc.DataDir, subDir, configFileName)
-	jsonBs, err := os.ReadFile(name)
+	// If all the TLS files are missing, assume this is intentional.
+	// Existing clusters do not receive any TLS certs.
+	if len(missing) == len(files) {
+		return nil
+	}
+
+	// If only some of the files are missing, something went wrong.
+	if len(missing) > 0 {
+		return fmt.Errorf("configuration files on disk are incomplete, missing: %v", missing)
+	}
+
+	cert, key, caCerts, err := loadCerts(dir)
+	if err != nil {
+		return fmt.Errorf("failed to load certs from disk: %w", err)
+	}
+
+	if err = validateTLSCerts(cert, key, caCerts); err != nil {
+		return fmt.Errorf("invalid certs on disk: %w", err)
+	}
+	return nil
+}
+
+func loadCerts(dir string) (cert, key string, caCerts []string, err error) {
+	certPEMBlock, err := os.ReadFile(filepath.Join(dir, certFileName))
+	if err != nil {
+		return "", "", nil, err
+	}
+	keyPEMBlock, err := os.ReadFile(filepath.Join(dir, keyFileName))
+	if err != nil {
+		return "", "", nil, err
+	}
+
+	caPEMs, err := os.ReadFile(filepath.Join(dir, caFileName))
+	if err != nil {
+		return "", "", nil, err
+	}
+	caCerts, err = splitCACerts(caPEMs)
 	if err != nil {
-		ui.Warn(fmt.Sprintf("failed to read local bootstrap config file, ignoring local files: %s", err))
-		return "", false
+		return "", "", nil, fmt.Errorf("failed to parse CA certs: %w", err)
+	}
+
+	return string(certPEMBlock), string(keyPEMBlock), caCerts, nil
+}
+
+// splitCACerts takes a list of concatenated PEM blocks and splits
+// them back up into strings. This is used because CACerts are written
+// into a single file, but validated individually.
+func splitCACerts(caPEMs []byte) ([]string, error) {
+	var out []string
+
+	for {
+		nextBlock, remaining := pem.Decode(caPEMs)
+		if nextBlock == nil {
+			break
+		}
+		if nextBlock.Type != "CERTIFICATE" {
+			return nil, fmt.Errorf("PEM-block should be CERTIFICATE type")
+		}
+
+		// Collect up to the start of the remaining bytes.
+		// We don't grab nextBlock.Bytes because it's not PEM encoded.
+		out = append(out, string(caPEMs[:len(caPEMs)-len(remaining)]))
+		caPEMs = remaining
 	}
 
-	// Check this looks non-empty at least
-	jsonStr := strings.TrimSpace(string(jsonBs))
-	// 50 is arbitrary but config containing the right secrets would always be
-	// bigger than this in JSON format so it is a reasonable test that this wasn't
-	// empty or just an empty JSON object or something.
-	if len(jsonStr) < 50 {
-		ui.Warn("ignoring incomplete local bootstrap config files")
-		return "", false
+	if len(out) == 0 {
+		return nil, errors.New("invalid CA certificate")
+	}
+	return out, nil
+}
+
+// validateTLSCerts checks that the CA cert, server cert, and key on disk are structurally valid.
+//
+// OPTIMIZE: This could be improved by returning an error if certs are expired or close to expiration.
+// However, that requires issuing new certs on bootstrap requests, since returning an error
+// would trigger a re-fetch from HCP.
+func validateTLSCerts(cert, key string, caCerts []string) error {
+	leaf, err := tls.X509KeyPair([]byte(cert), []byte(key))
+	if err != nil {
+		return errors.New("invalid server certificate or key")
+	}
+	_, err = x509.ParseCertificate(leaf.Certificate[0])
+	if err != nil {
+		return errors.New("invalid server certificate")
 	}
 
-	// TODO we could parse the certificates and check they are still valid here
-	// and force a reload if not. We could also attempt to parse config and check
-	// it's all valid just in case the local config was really old and has
-	// deprecated fields or something?
-	return jsonStr, true
+	for _, caCert := range caCerts {
+		_, err = connect.ParseCert(caCert)
+		if err != nil {
+			return errors.New("invalid CA certificate")
+		}
+	}
+	return nil
 }
diff --git a/agent/hcp/bootstrap/bootstrap_test.go b/agent/hcp/bootstrap/bootstrap_test.go
new file mode 100644
index 000000000000..5d00d3372d08
--- /dev/null
+++ b/agent/hcp/bootstrap/bootstrap_test.go
@@ -0,0 +1,467 @@
+package bootstrap
+
+import (
+	"context"
+	"crypto/tls"
+	"crypto/x509"
+	"fmt"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/hashicorp/consul/agent/config"
+	"github.com/hashicorp/consul/agent/hcp"
+	"github.com/hashicorp/consul/lib"
+	"github.com/hashicorp/consul/tlsutil"
+	"github.com/hashicorp/go-uuid"
+	"github.com/mitchellh/cli"
+	"github.com/stretchr/testify/require"
+)
+
+func TestBootstrapConfigLoader(t *testing.T) {
+	baseLoader := func(source config.Source) (config.LoadResult, error) {
+		return config.Load(config.LoadOpts{
+			DefaultConfig: source,
+			HCL: []string{
+				`server = true`,
+				`bind_addr = "127.0.0.1"`,
+				`data_dir = "/tmp/consul-data"`,
+			},
+		})
+	}
+
+	bootstrapLoader := func(source config.Source) (config.LoadResult, error) {
+		return bootstrapConfigLoader(baseLoader, &RawBootstrapConfig{
+			ConfigJSON:      `{"bootstrap_expect": 8}`,
+			ManagementToken: "test-token",
+		})(source)
+	}
+
+	result, err := bootstrapLoader(nil)
+	require.NoError(t, err)
+
+	// bootstrap_expect and management token are injected from bootstrap config received from HCP.
+	require.Equal(t, 8, result.RuntimeConfig.BootstrapExpect)
+	require.Equal(t, "test-token", result.RuntimeConfig.Cloud.ManagementToken)
+
+	// Response header is always injected from a constant.
+	require.Equal(t, "x-consul-default-acl-policy", result.RuntimeConfig.HTTPResponseHeaders[accessControlHeaderName])
+}
+
+func Test_finalizeRuntimeConfig(t *testing.T) {
+	type testCase struct {
+		rc       *config.RuntimeConfig
+		cfg      *RawBootstrapConfig
+		verifyFn func(t *testing.T, rc *config.RuntimeConfig)
+	}
+	run := func(t *testing.T, tc testCase) {
+		finalizeRuntimeConfig(tc.rc, tc.cfg)
+		tc.verifyFn(t, tc.rc)
+	}
+
+	tt := map[string]testCase{
+		"set header if not present": {
+			rc: &config.RuntimeConfig{},
+			cfg: &RawBootstrapConfig{
+				ManagementToken: "test-token",
+			},
+			verifyFn: func(t *testing.T, rc *config.RuntimeConfig) {
+				require.Equal(t, "test-token", rc.Cloud.ManagementToken)
+				require.Equal(t, "x-consul-default-acl-policy", rc.HTTPResponseHeaders[accessControlHeaderName])
+			},
+		},
+		"append to header if present": {
+			rc: &config.RuntimeConfig{
+				HTTPResponseHeaders: map[string]string{
+					accessControlHeaderName: "Content-Encoding",
+				},
+			},
+			cfg: &RawBootstrapConfig{
+				ManagementToken: "test-token",
+			},
+			verifyFn: func(t *testing.T, rc *config.RuntimeConfig) {
+				require.Equal(t, "test-token", rc.Cloud.ManagementToken)
+				require.Equal(t, "Content-Encoding,x-consul-default-acl-policy", rc.HTTPResponseHeaders[accessControlHeaderName])
+			},
+		},
+	}
+
+	for name, tc := range tt {
+		t.Run(name, func(t *testing.T) {
+			run(t, tc)
+		})
+	}
+}
+
+func boolPtr(value bool) *bool {
+	return &value
+}
+
+func TestLoadConfig_Persistence(t *testing.T) {
+	type testCase struct {
+		// resourceID is the HCP resource ID. If set, a server is considered to be cloud-enabled.
+		resourceID string
+
+		// devMode indicates whether the loader should not have a data directory.
+		devMode bool
+
+		// verifyFn issues case-specific assertions.
+		verifyFn func(t *testing.T, rc *config.RuntimeConfig)
+	}
+
+	run := func(t *testing.T, tc testCase) {
+		dir, err := os.MkdirTemp(os.TempDir(), "bootstrap-test-")
+		require.NoError(t, err)
+		t.Cleanup(func() { os.RemoveAll(dir) })
+
+		s := hcp.NewMockHCPServer()
+		s.AddEndpoint(TestEndpoint())
+
+		// Use an HTTPS server since that's what the HCP SDK expects for auth.
+		srv := httptest.NewTLSServer(s)
+		defer srv.Close()
+
+		caCert, err := x509.ParseCertificate(srv.TLS.Certificates[0].Certificate[0])
+		require.NoError(t, err)
+
+		pool := x509.NewCertPool()
+		pool.AddCert(caCert)
+		clientTLS := &tls.Config{RootCAs: pool}
+
+		baseOpts := config.LoadOpts{
+			HCL: []string{
+				`server = true`,
+				`bind_addr = "127.0.0.1"`,
+				fmt.Sprintf(`http_config = { response_headers = { %s = "Content-Encoding" } }`, accessControlHeaderName),
+				fmt.Sprintf(`cloud { client_id="test" client_secret="test" hostname=%q auth_url=%q resource_id=%q }`,
+					srv.Listener.Addr().String(), srv.URL, tc.resourceID),
+			},
+		}
+		if tc.devMode {
+			baseOpts.DevMode = boolPtr(true)
+		} else {
+			baseOpts.HCL = append(baseOpts.HCL, fmt.Sprintf(`data_dir = %q`, dir))
+		}
+
+		baseLoader := func(source config.Source) (config.LoadResult, error) {
+			baseOpts.DefaultConfig = source
+			return config.Load(baseOpts)
+		}
+
+		ui := cli.NewMockUi()
+
+		// Load initial config to check whether bootstrapping from HCP is enabled.
+		initial, err := baseLoader(nil)
+		require.NoError(t, err)
+
+		// Override the client TLS config so that the test server can be trusted.
+		initial.RuntimeConfig.Cloud.WithTLSConfig(clientTLS)
+		client, err := hcp.NewClient(initial.RuntimeConfig.Cloud)
+		require.NoError(t, err)
+
+		loader, err := LoadConfig(context.Background(), client, initial.RuntimeConfig.DataDir, baseLoader, ui)
+		require.NoError(t, err)
+
+		// Load the agent config with the potentially wrapped loader.
+		fromRemote, err := loader(nil)
+		require.NoError(t, err)
+
+		// HCP-enabled cases should fetch from HCP on the first run of LoadConfig.
+		require.Contains(t, ui.OutputWriter.String(), "Fetching configuration from HCP")
+
+		// Run case-specific verification.
+		tc.verifyFn(t, fromRemote.RuntimeConfig)
+
+		require.Empty(t, fromRemote.RuntimeConfig.ACLInitialManagementToken,
+			"initial_management token should have been sanitized")
+
+		if tc.devMode {
+			// Re-running the bootstrap func below isn't relevant to dev mode
+			// since they don't have a data directory to load data from.
+			return
+		}
+
+		// Run LoadConfig again to exercise the logic of loading config from disk.
+		loader, err = LoadConfig(context.Background(), client, initial.RuntimeConfig.DataDir, baseLoader, ui)
+		require.NoError(t, err)
+
+		fromDisk, err := loader(nil)
+		require.NoError(t, err)
+
+		// HCP-enabled cases should fetch from disk on the second run.
+		require.Contains(t, ui.OutputWriter.String(), "Loaded HCP configuration from local disk")
+
+		// Config loaded from disk should be the same as the one that was initially fetched from the HCP servers.
+		require.Equal(t, fromRemote.RuntimeConfig, fromDisk.RuntimeConfig)
+	}
+
+	tt := map[string]testCase{
+		"dev mode": {
+			devMode: true,
+
+			resourceID: "organization/0b9de9a3-8403-4ca6-aba8-fca752f42100/" +
+				"project/0b9de9a3-8403-4ca6-aba8-fca752f42100/" +
+				"consul.cluster/new-cluster-id",
+
+			verifyFn: func(t *testing.T, rc *config.RuntimeConfig) {
+				require.Empty(t, rc.DataDir)
+
+				// Dev mode should have persisted certs since they can't be inlined.
+				require.NotEmpty(t, rc.TLS.HTTPS.CertFile)
+				require.NotEmpty(t, rc.TLS.HTTPS.KeyFile)
+				require.NotEmpty(t, rc.TLS.HTTPS.CAFile)
+
+				// Find the temporary directory they got stored in.
+				dir := filepath.Dir(rc.TLS.HTTPS.CertFile)
+
+				// Ensure we only stored the TLS materials.
+				entries, err := os.ReadDir(dir)
+				require.NoError(t, err)
+				require.Len(t, entries, 3)
+
+				haveFiles := make([]string, 3)
+				for i, entry := range entries {
+					haveFiles[i] = entry.Name()
+				}
+
+				wantFiles := []string{caFileName, certFileName, keyFileName}
+				require.ElementsMatch(t, wantFiles, haveFiles)
+			},
+		},
+		"new cluster": {
+			resourceID: "organization/0b9de9a3-8403-4ca6-aba8-fca752f42100/" +
+				"project/0b9de9a3-8403-4ca6-aba8-fca752f42100/" +
+				"consul.cluster/new-cluster-id",
+
+			// New clusters should have received and persisted the whole suite of config.
+			verifyFn: func(t *testing.T, rc *config.RuntimeConfig) {
+				dir := filepath.Join(rc.DataDir, subDir)
+
+				entries, err := os.ReadDir(dir)
+				require.NoError(t, err)
+				require.Len(t, entries, 6)
+
+				files := []string{
+					filepath.Join(dir, configFileName),
+					filepath.Join(dir, caFileName),
+					filepath.Join(dir, certFileName),
+					filepath.Join(dir, keyFileName),
+					filepath.Join(dir, tokenFileName),
+					filepath.Join(dir, successFileName),
+				}
+				for _, name := range files {
+					_, err := os.Stat(name)
+					require.NoError(t, err)
+				}
+
+				require.Equal(t, filepath.Join(dir, certFileName), rc.TLS.HTTPS.CertFile)
+				require.Equal(t, filepath.Join(dir, keyFileName), rc.TLS.HTTPS.KeyFile)
+				require.Equal(t, filepath.Join(dir, caFileName), rc.TLS.HTTPS.CAFile)
+
+				cert, key, caCerts, err := loadCerts(dir)
+				require.NoError(t, err)
+
+				require.NoError(t, validateTLSCerts(cert, key, caCerts))
+			},
+		},
+		"existing cluster": {
+			resourceID: "organization/0b9de9a3-8403-4ca6-aba8-fca752f42100/" +
+				"project/0b9de9a3-8403-4ca6-aba8-fca752f42100/" +
+				"consul.cluster/" + TestExistingClusterID,
+
+			// Existing clusters should have only received and persisted the management token.
+			verifyFn: func(t *testing.T, rc *config.RuntimeConfig) {
+				dir := filepath.Join(rc.DataDir, subDir)
+
+				entries, err := os.ReadDir(dir)
+				require.NoError(t, err)
+				require.Len(t, entries, 3)
+
+				files := []string{
+					filepath.Join(dir, tokenFileName),
+					filepath.Join(dir, successFileName),
+					filepath.Join(dir, configFileName),
+				}
+				for _, name := range files {
+					_, err := os.Stat(name)
+					require.NoError(t, err)
+				}
+			},
+		},
+	}
+
+	for name, tc := range tt {
+		t.Run(name, func(t *testing.T) {
+			run(t, tc)
+		})
+	}
+}
+
+func Test_loadPersistedBootstrapConfig(t *testing.T) {
+	type expect struct {
+		loaded  bool
+		warning string
+	}
+	type testCase struct {
+		existingCluster bool
+		mutateFn        func(t *testing.T, dir string)
+		expect          expect
+	}
+
+	run := func(t *testing.T, tc testCase) {
+		dataDir, err := os.MkdirTemp(os.TempDir(), "load-bootstrap-test-")
+		require.NoError(t, err)
+		t.Cleanup(func() { os.RemoveAll(dataDir) })
+
+		dir := filepath.Join(dataDir, subDir)
+
+		// Do some common setup as if we received config from HCP and persisted it to disk.
+		require.NoError(t, lib.EnsurePath(dir, true))
+		require.NoError(t, persistSucessMarker(dir))
+
+		if !tc.existingCluster {
+			caCert, caKey, err := tlsutil.GenerateCA(tlsutil.CAOpts{})
+			require.NoError(t, err)
+
+			serverCert, serverKey, err := testLeaf(caCert, caKey)
+			require.NoError(t, err)
+			require.NoError(t, persistTLSCerts(dir, serverCert, serverKey, []string{caCert}))
+
+			cfgJSON := `{"bootstrap_expect": 8}`
+			require.NoError(t, persistBootstrapConfig(dir, cfgJSON))
+		}
+
+		token, err := uuid.GenerateUUID()
+		require.NoError(t, err)
+		require.NoError(t, persistManagementToken(dir, token))
+
+		// Optionally mutate the persisted data to trigger errors while loading.
+		if tc.mutateFn != nil {
+			tc.mutateFn(t, dir)
+		}
+
+		ui := cli.NewMockUi()
+		cfg, loaded := loadPersistedBootstrapConfig(dataDir, ui)
+		require.Equal(t, tc.expect.loaded, loaded, ui.ErrorWriter.String())
+		if loaded {
+			require.Equal(t, token, cfg.ManagementToken)
+			require.Empty(t, ui.ErrorWriter.String())
+
+		} else {
+			require.Nil(t, cfg)
+			require.Contains(t, ui.ErrorWriter.String(), tc.expect.warning)
+		}
+	}
+
+	tt := map[string]testCase{
+		"existing cluster with valid files": {
+			existingCluster: true,
+			// Don't mutate, files from setup are valid.
+			mutateFn: nil,
+			expect: expect{
+				loaded:  true,
+				warning: "",
+			},
+		},
+		"existing cluster missing token": {
+			existingCluster: true,
+			mutateFn: func(t *testing.T, dir string) {
+				// Remove the token file while leaving the existing cluster marker.
+				require.NoError(t, os.Remove(filepath.Join(dir, tokenFileName)))
+			},
+			expect: expect{
+				loaded:  false,
+				warning: "configuration files on disk are incomplete",
+			},
+		},
+		"existing cluster no files": {
+			existingCluster: true,
+			mutateFn: func(t *testing.T, dir string) {
+				// Remove all files
+				require.NoError(t, os.RemoveAll(dir))
+			},
+			expect: expect{
+				loaded: false,
+				// No warnings since we assume we need to fetch config from HCP for the first time.
+				warning: "",
+			},
+		},
+		"new cluster with valid files": {
+			// Don't mutate, files from setup are valid.
+			mutateFn: nil,
+			expect: expect{
+				loaded:  true,
+				warning: "",
+			},
+		},
+		"new cluster some files": {
+			mutateFn: func(t *testing.T, dir string) {
+				// Remove one of the required files
+				require.NoError(t, os.Remove(filepath.Join(dir, certFileName)))
+			},
+			expect: expect{
+				loaded:  false,
+				warning: "configuration files on disk are incomplete",
+			},
+		},
+		"new cluster no files": {
+			mutateFn: func(t *testing.T, dir string) {
+				// Remove all files
+				require.NoError(t, os.RemoveAll(dir))
+			},
+			expect: expect{
+				loaded: false,
+				// No warnings since we assume we need to fetch config from HCP for the first time.
+				warning: "",
+			},
+		},
+		"new cluster invalid cert": {
+			mutateFn: func(t *testing.T, dir string) {
+				name := filepath.Join(dir, certFileName)
+				require.NoError(t, os.WriteFile(name, []byte("not-a-cert"), 0600))
+			},
+			expect: expect{
+				loaded:  false,
+				warning: "invalid server certificate",
+			},
+		},
+		"new cluster invalid CA": {
+			mutateFn: func(t *testing.T, dir string) {
+				name := filepath.Join(dir, caFileName)
+				require.NoError(t, os.WriteFile(name, []byte("not-a-ca-cert"), 0600))
+			},
+			expect: expect{
+				loaded:  false,
+				warning: "invalid CA certificate",
+			},
+		},
+		"new cluster invalid config flag": {
+			mutateFn: func(t *testing.T, dir string) {
+				name := filepath.Join(dir, configFileName)
+				require.NoError(t, os.WriteFile(name, []byte(`{"not_a_consul_agent_config_field" = "zap"}`), 0600))
+			},
+			expect: expect{
+				loaded:  false,
+				warning: "failed to parse local bootstrap config",
+			},
+		},
+		"existing cluster invalid token": {
+			existingCluster: true,
+			mutateFn: func(t *testing.T, dir string) {
+				name := filepath.Join(dir, tokenFileName)
+				require.NoError(t, os.WriteFile(name, []byte("not-a-uuid"), 0600))
+			},
+			expect: expect{
+				loaded:  false,
+				warning: "is not a valid UUID",
+			},
+		},
+	}
+
+	for name, tc := range tt {
+		t.Run(name, func(t *testing.T) {
+			run(t, tc)
+		})
+	}
+}
diff --git a/agent/hcp/bootstrap/testing.go b/agent/hcp/bootstrap/testing.go
index ef8941f08be1..a10a5d2bc8ad 100644
--- a/agent/hcp/bootstrap/testing.go
+++ b/agent/hcp/bootstrap/testing.go
@@ -83,7 +83,7 @@ func generateClusterData(cluster resource.Resource) (gnmmod.HashicorpCloudGlobal
 		if err != nil {
 			return resp, err
 		}
-		resp.Bootstrap.ConsulConfig = ""
+		resp.Bootstrap.ConsulConfig = "{}"
 		resp.Bootstrap.ManagementToken = token
 		return resp, nil
 	}
@@ -157,7 +157,8 @@ func generateClusterData(cluster resource.Resource) (gnmmod.HashicorpCloudGlobal
 		"tokens": map[string]interface{}{
 			// Also setup the server's own agent token to be the management token so it has
 			// permission to register itself.
-			"agent": token,
+			"agent":              token,
+			"initial_management": token,
 		},
 		"default_policy":           "deny",
 		"enabled":                  true,
diff --git a/agent/hcp/config/config.go b/agent/hcp/config/config.go
index d087c67fa972..a6d4c31979db 100644
--- a/agent/hcp/config/config.go
+++ b/agent/hcp/config/config.go
@@ -18,22 +18,33 @@ type CloudConfig struct {
 	AuthURL      string
 	ScadaAddress string
 
-	// internal
+	// Management token used by HCP management plane.
+	// Cannot be set via config files.
 	ManagementToken string
+
+	// TlsConfig for testing.
+	TLSConfig *tls.Config
+}
+
+func (c *CloudConfig) WithTLSConfig(cfg *tls.Config) {
+	c.TLSConfig = cfg
 }
 
 func (c *CloudConfig) HCPConfig(opts ...hcpcfg.HCPConfigOption) (hcpcfg.HCPConfig, error) {
+	if c.TLSConfig == nil {
+		c.TLSConfig = &tls.Config{}
+	}
 	if c.ClientID != "" && c.ClientSecret != "" {
 		opts = append(opts, hcpcfg.WithClientCredentials(c.ClientID, c.ClientSecret))
 	}
 	if c.AuthURL != "" {
-		opts = append(opts, hcpcfg.WithAuth(c.AuthURL, &tls.Config{}))
+		opts = append(opts, hcpcfg.WithAuth(c.AuthURL, c.TLSConfig))
 	}
 	if c.Hostname != "" {
-		opts = append(opts, hcpcfg.WithAPI(c.Hostname, &tls.Config{}))
+		opts = append(opts, hcpcfg.WithAPI(c.Hostname, c.TLSConfig))
 	}
 	if c.ScadaAddress != "" {
-		opts = append(opts, hcpcfg.WithSCADA(c.ScadaAddress, &tls.Config{}))
+		opts = append(opts, hcpcfg.WithSCADA(c.ScadaAddress, c.TLSConfig))
 	}
 	opts = append(opts, hcpcfg.FromEnv())
 	return hcpcfg.NewHCPConfig(opts...)
diff --git a/agent/hcp/scada/mock_Provider.go b/agent/hcp/scada/mock_Provider.go
index 251178095bc2..b9a0fd2d4967 100644
--- a/agent/hcp/scada/mock_Provider.go
+++ b/agent/hcp/scada/mock_Provider.go
@@ -1,4 +1,4 @@
-// Code generated by mockery v2.15.0. DO NOT EDIT.
+// Code generated by mockery v2.20.0. DO NOT EDIT.
 
 package scada
 
@@ -7,6 +7,8 @@ import (
 
 	mock "github.com/stretchr/testify/mock"
 
+	provider "github.com/hashicorp/hcp-scada-provider"
+
 	time "time"
 )
 
@@ -23,6 +25,98 @@ func (_m *MockProvider) EXPECT() *MockProvider_Expecter {
 	return &MockProvider_Expecter{mock: &_m.Mock}
 }
 
+// AddMeta provides a mock function with given fields: _a0
+func (_m *MockProvider) AddMeta(_a0 ...provider.Meta) {
+	_va := make([]interface{}, len(_a0))
+	for _i := range _a0 {
+		_va[_i] = _a0[_i]
+	}
+	var _ca []interface{}
+	_ca = append(_ca, _va...)
+	_m.Called(_ca...)
+}
+
+// MockProvider_AddMeta_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'AddMeta'
+type MockProvider_AddMeta_Call struct {
+	*mock.Call
+}
+
+// AddMeta is a helper method to define mock.On call
+//   - _a0 ...provider.Meta
+func (_e *MockProvider_Expecter) AddMeta(_a0 ...interface{}) *MockProvider_AddMeta_Call {
+	return &MockProvider_AddMeta_Call{Call: _e.mock.On("AddMeta",
+		append([]interface{}{}, _a0...)...)}
+}
+
+func (_c *MockProvider_AddMeta_Call) Run(run func(_a0 ...provider.Meta)) *MockProvider_AddMeta_Call {
+	_c.Call.Run(func(args mock.Arguments) {
+		variadicArgs := make([]provider.Meta, len(args)-0)
+		for i, a := range args[0:] {
+			if a != nil {
+				variadicArgs[i] = a.(provider.Meta)
+			}
+		}
+		run(variadicArgs...)
+	})
+	return _c
+}
+
+func (_c *MockProvider_AddMeta_Call) Return() *MockProvider_AddMeta_Call {
+	_c.Call.Return()
+	return _c
+}
+
+func (_c *MockProvider_AddMeta_Call) RunAndReturn(run func(...provider.Meta)) *MockProvider_AddMeta_Call {
+	_c.Call.Return(run)
+	return _c
+}
+
+// DeleteMeta provides a mock function with given fields: _a0
+func (_m *MockProvider) DeleteMeta(_a0 ...string) {
+	_va := make([]interface{}, len(_a0))
+	for _i := range _a0 {
+		_va[_i] = _a0[_i]
+	}
+	var _ca []interface{}
+	_ca = append(_ca, _va...)
+	_m.Called(_ca...)
+}
+
+// MockProvider_DeleteMeta_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'DeleteMeta'
+type MockProvider_DeleteMeta_Call struct {
+	*mock.Call
+}
+
+// DeleteMeta is a helper method to define mock.On call
+//   - _a0 ...string
+func (_e *MockProvider_Expecter) DeleteMeta(_a0 ...interface{}) *MockProvider_DeleteMeta_Call {
+	return &MockProvider_DeleteMeta_Call{Call: _e.mock.On("DeleteMeta",
+		append([]interface{}{}, _a0...)...)}
+}
+
+func (_c *MockProvider_DeleteMeta_Call) Run(run func(_a0 ...string)) *MockProvider_DeleteMeta_Call {
+	_c.Call.Run(func(args mock.Arguments) {
+		variadicArgs := make([]string, len(args)-0)
+		for i, a := range args[0:] {
+			if a != nil {
+				variadicArgs[i] = a.(string)
+			}
+		}
+		run(variadicArgs...)
+	})
+	return _c
+}
+
+func (_c *MockProvider_DeleteMeta_Call) Return() *MockProvider_DeleteMeta_Call {
+	_c.Call.Return()
+	return _c
+}
+
+func (_c *MockProvider_DeleteMeta_Call) RunAndReturn(run func(...string)) *MockProvider_DeleteMeta_Call {
+	_c.Call.Return(run)
+	return _c
+}
+
 // GetMeta provides a mock function with given fields:
 func (_m *MockProvider) GetMeta() map[string]string {
 	ret := _m.Called()
@@ -61,18 +155,26 @@ func (_c *MockProvider_GetMeta_Call) Return(_a0 map[string]string) *MockProvider
 	return _c
 }
 
+func (_c *MockProvider_GetMeta_Call) RunAndReturn(run func() map[string]string) *MockProvider_GetMeta_Call {
+	_c.Call.Return(run)
+	return _c
+}
+
 // LastError provides a mock function with given fields:
 func (_m *MockProvider) LastError() (time.Time, error) {
 	ret := _m.Called()
 
 	var r0 time.Time
+	var r1 error
+	if rf, ok := ret.Get(0).(func() (time.Time, error)); ok {
+		return rf()
+	}
 	if rf, ok := ret.Get(0).(func() time.Time); ok {
 		r0 = rf()
 	} else {
 		r0 = ret.Get(0).(time.Time)
 	}
 
-	var r1 error
 	if rf, ok := ret.Get(1).(func() error); ok {
 		r1 = rf()
 	} else {
@@ -104,11 +206,20 @@ func (_c *MockProvider_LastError_Call) Return(_a0 time.Time, _a1 error) *MockPro
 	return _c
 }
 
+func (_c *MockProvider_LastError_Call) RunAndReturn(run func() (time.Time, error)) *MockProvider_LastError_Call {
+	_c.Call.Return(run)
+	return _c
+}
+
 // Listen provides a mock function with given fields: capability
 func (_m *MockProvider) Listen(capability string) (net.Listener, error) {
 	ret := _m.Called(capability)
 
 	var r0 net.Listener
+	var r1 error
+	if rf, ok := ret.Get(0).(func(string) (net.Listener, error)); ok {
+		return rf(capability)
+	}
 	if rf, ok := ret.Get(0).(func(string) net.Listener); ok {
 		r0 = rf(capability)
 	} else {
@@ -117,7 +228,6 @@ func (_m *MockProvider) Listen(capability string) (net.Listener, error) {
 		}
 	}
 
-	var r1 error
 	if rf, ok := ret.Get(1).(func(string) error); ok {
 		r1 = rf(capability)
 	} else {
@@ -150,6 +260,11 @@ func (_c *MockProvider_Listen_Call) Return(_a0 net.Listener, _a1 error) *MockPro
 	return _c
 }
 
+func (_c *MockProvider_Listen_Call) RunAndReturn(run func(string) (net.Listener, error)) *MockProvider_Listen_Call {
+	_c.Call.Return(run)
+	return _c
+}
+
 // SessionStatus provides a mock function with given fields:
 func (_m *MockProvider) SessionStatus() string {
 	ret := _m.Called()
@@ -186,6 +301,11 @@ func (_c *MockProvider_SessionStatus_Call) Return(_a0 string) *MockProvider_Sess
 	return _c
 }
 
+func (_c *MockProvider_SessionStatus_Call) RunAndReturn(run func() string) *MockProvider_SessionStatus_Call {
+	_c.Call.Return(run)
+	return _c
+}
+
 // Start provides a mock function with given fields:
 func (_m *MockProvider) Start() error {
 	ret := _m.Called()
@@ -222,6 +342,11 @@ func (_c *MockProvider_Start_Call) Return(_a0 error) *MockProvider_Start_Call {
 	return _c
 }
 
+func (_c *MockProvider_Start_Call) RunAndReturn(run func() error) *MockProvider_Start_Call {
+	_c.Call.Return(run)
+	return _c
+}
+
 // Stop provides a mock function with given fields:
 func (_m *MockProvider) Stop() error {
 	ret := _m.Called()
@@ -258,6 +383,11 @@ func (_c *MockProvider_Stop_Call) Return(_a0 error) *MockProvider_Stop_Call {
 	return _c
 }
 
+func (_c *MockProvider_Stop_Call) RunAndReturn(run func() error) *MockProvider_Stop_Call {
+	_c.Call.Return(run)
+	return _c
+}
+
 // UpdateMeta provides a mock function with given fields: _a0
 func (_m *MockProvider) UpdateMeta(_a0 map[string]string) {
 	_m.Called(_a0)
@@ -286,6 +416,11 @@ func (_c *MockProvider_UpdateMeta_Call) Return() *MockProvider_UpdateMeta_Call {
 	return _c
 }
 
+func (_c *MockProvider_UpdateMeta_Call) RunAndReturn(run func(map[string]string)) *MockProvider_UpdateMeta_Call {
+	_c.Call.Return(run)
+	return _c
+}
+
 type mockConstructorTestingTNewMockProvider interface {
 	mock.TestingT
 	Cleanup(func())
diff --git a/command/agent/agent.go b/command/agent/agent.go
index 356faf30d3f4..a4784f626f45 100644
--- a/command/agent/agent.go
+++ b/command/agent/agent.go
@@ -15,6 +15,7 @@ import (
 	"syscall"
 	"time"
 
+	"github.com/hashicorp/consul/agent/hcp"
 	"github.com/hashicorp/go-checkpoint"
 	"github.com/hashicorp/go-hclog"
 	mcli "github.com/mitchellh/cli"
@@ -159,14 +160,28 @@ func (c *cmd) run(args []string) int {
 	go handleStartupSignals(ctx, cancel, signalCh, suLogger)
 
 	// See if we need to bootstrap config from HCP before we go any further with
-	// agent startup. We override loader with the one returned as it may be
-	// modified to include HCP-provided config.
-	var err error
-	_, loader, err = hcpbootstrap.MaybeBootstrap(ctx, loader, ui)
+	// agent startup. First do a preliminary load of agent configuration using the given loader.
+	// This is just to peek whether bootstrapping from HCP is enabled. The result is discarded
+	// on the call to agent.NewBaseDeps so that the wrapped loader takes effect.
+	res, err := loader(nil)
 	if err != nil {
 		ui.Error(err.Error())
 		return 1
 	}
+	if res.RuntimeConfig.IsCloudEnabled() {
+		client, err := hcp.NewClient(res.RuntimeConfig.Cloud)
+		if err != nil {
+			ui.Error("error building HCP HTTP client: " + err.Error())
+			return 1
+		}
+
+		// We override loader with the one returned as it was modified to include HCP-provided config.
+		loader, err = hcpbootstrap.LoadConfig(ctx, client, res.RuntimeConfig.DataDir, loader, ui)
+		if err != nil {
+			ui.Error(err.Error())
+			return 1
+		}
+	}
 
 	bd, err := agent.NewBaseDeps(loader, logGate, nil)
 	if err != nil {
diff --git a/go.mod b/go.mod
index 4e4aa691eaac..333b699300f2 100644
--- a/go.mod
+++ b/go.mod
@@ -60,7 +60,7 @@ require (
 	github.com/hashicorp/go-version v1.2.1
 	github.com/hashicorp/golang-lru v0.5.4
 	github.com/hashicorp/hcl v1.0.0
-	github.com/hashicorp/hcp-scada-provider v0.2.0
+	github.com/hashicorp/hcp-scada-provider v0.2.3
 	github.com/hashicorp/hcp-sdk-go v0.40.1-0.20230404193545-846aea419cd1
 	github.com/hashicorp/hil v0.0.0-20200423225030-a18a1cd20038
 	github.com/hashicorp/memberlist v0.5.0
diff --git a/go.sum b/go.sum
index ebe36026d017..2bb892bc9ca3 100644
--- a/go.sum
+++ b/go.sum
@@ -602,8 +602,8 @@ github.com/hashicorp/golang-lru v0.5.4 h1:YDjusn29QI/Das2iO9M0BHnIbxPeyuCHsjMW+l
 github.com/hashicorp/golang-lru v0.5.4/go.mod h1:iADmTwqILo4mZ8BN3D2Q6+9jd8WM5uGBxy+E8yxSoD4=
 github.com/hashicorp/hcl v1.0.0 h1:0Anlzjpi4vEasTeNFn2mLJgTSwt0+6sfsiTG8qcWGx4=
 github.com/hashicorp/hcl v1.0.0/go.mod h1:E5yfLk+7swimpb2L/Alb/PJmXilQ/rhwaUYs4T20WEQ=
-github.com/hashicorp/hcp-scada-provider v0.2.0 h1:iD3Y+c7LTdjeaWKHq/ym6ahEdSL1R+9GHvKWBb4t+aM=
-github.com/hashicorp/hcp-scada-provider v0.2.0/go.mod h1:Q0WpS2RyhBKOPD4X/8oW7AJe7jA2HXB09EwDzwRTao0=
+github.com/hashicorp/hcp-scada-provider v0.2.3 h1:AarYR+/Pcv+cMvPdAlb92uOBmZfEH6ny4+DT+4NY2VQ=
+github.com/hashicorp/hcp-scada-provider v0.2.3/go.mod h1:ZFTgGwkzNv99PLQjTsulzaCplCzOTBh0IUQsPKzrQFo=
 github.com/hashicorp/hcp-sdk-go v0.40.1-0.20230404193545-846aea419cd1 h1:C1des4/oIeUqQJVUWypnZth19Kg+k01q+V59OVNMB+Q=
 github.com/hashicorp/hcp-sdk-go v0.40.1-0.20230404193545-846aea419cd1/go.mod h1:hZqky4HEzsKwvLOt4QJlZUrjeQmb4UCZUhDP2HyQFfc=
 github.com/hashicorp/hil v0.0.0-20200423225030-a18a1cd20038 h1:n9J0rwVWXDpNd5iZnwY7w4WZyq53/rROeI7OVvLW8Ok=