diff --git a/changelog/pending/20230128--backend-filestate--the-filestate-backend-now-supports-project-scoped-stacks.yaml b/changelog/pending/20230128--backend-filestate--the-filestate-backend-now-supports-project-scoped-stacks.yaml
new file mode 100644
index 000000000000..8c85a6b9a308
--- /dev/null
+++ b/changelog/pending/20230128--backend-filestate--the-filestate-backend-now-supports-project-scoped-stacks.yaml
@@ -0,0 +1,9 @@
+changes:
+- type: feat
+  scope: backend/filestate
+  description: |
+    Add support for project-scoped stacks.
+    Newly initialized storage will automatically use this mode.
+    Set PULUMI_SELF_MANAGED_STATE_LEGACY_LAYOUT=1 to opt-out of this.
+    This mode needs write access to the root of the .pulumi directory;
+    if you're using a cloud storage, be sure to update your ACLs.
diff --git a/changelog/pending/20230328--backend-filestate--print-a-warning-if-a-project-scoped-storage-has-non-project-stacks-in-it.yaml b/changelog/pending/20230328--backend-filestate--print-a-warning-if-a-project-scoped-storage-has-non-project-stacks-in-it.yaml
new file mode 100644
index 000000000000..bc82bf5c1f97
--- /dev/null
+++ b/changelog/pending/20230328--backend-filestate--print-a-warning-if-a-project-scoped-storage-has-non-project-stacks-in-it.yaml
@@ -0,0 +1,6 @@
+changes:
+- type: chore
+  scope: backend/filestate
+  description: |
+    Print a warning if a project-scoped storage has non-project stacks in it.
+    Disable this warning by setting PULUMI_SELF_MANAGED_STATE_NO_LEGACY_WARNING=1.
diff --git a/pkg/backend/backend.go b/pkg/backend/backend.go
index b8cf8068ee12..ff8c5f90c22e 100644
--- a/pkg/backend/backend.go
+++ b/pkg/backend/backend.go
@@ -74,7 +74,7 @@ type StackReference interface {
 	// but that information is not part of the StackName() we pass to the engine.
 	Name() tokens.Name
 
-	// Fully qualified name of the stack.
+	// Fully qualified name of the stack, including any organization, project, or other information.
 	FullyQualifiedName() tokens.QName
 }
 
diff --git a/pkg/backend/filestate/backend.go b/pkg/backend/filestate/backend.go
index d029cbb477fb..3e4e81312385 100644
--- a/pkg/backend/filestate/backend.go
+++ b/pkg/backend/filestate/backend.go
@@ -53,6 +53,7 @@ import (
 	"github.com/pulumi/pulumi/sdk/v3/go/common/diag/colors"
 	sdkDisplay "github.com/pulumi/pulumi/sdk/v3/go/common/display"
 	"github.com/pulumi/pulumi/sdk/v3/go/common/encoding"
+	"github.com/pulumi/pulumi/sdk/v3/go/common/env"
 	"github.com/pulumi/pulumi/sdk/v3/go/common/resource/config"
 	"github.com/pulumi/pulumi/sdk/v3/go/common/tokens"
 	"github.com/pulumi/pulumi/sdk/v3/go/common/util/cmdutil"
@@ -65,6 +66,29 @@ import (
 // to enable gzip compression when using the filestate backend.
 const PulumiFilestateGzipEnvVar = "PULUMI_SELF_MANAGED_STATE_GZIP"
 
+// TODO[pulumi/pulumi#12539]:
+// This section contains names of environment variables
+// that affect the behavior of the backend.
+//
+// These must all be registered in common/env so that they're available
+// with the 'pulumi env' command.
+// However, we don't currently use env.Value() to access their values
+// because it prevents us from overriding the definition of os.Getenv
+// in tests.
+var (
+	// PulumiFilestateNoLegacyWarningEnvVar is an env var that must be truthy
+	// to disable the warning printed by the filestate backend
+	// when it detects that the state has both, project-scoped and legacy stacks.
+	PulumiFilestateNoLegacyWarningEnvVar = env.SelfManagedStateNoLegacyWarning.Var().Name()
+
+	// PulumiFilestateLegacyLayoutEnvVar is the name of an environment variable
+	// that can be set to force the use of the legacy layout
+	// when initializing an empty bucket for filestate.
+	//
+	// This opt-out is intended to be removed in a future release.
+	PulumiFilestateLegacyLayoutEnvVar = env.SelfManagedStateLegacyLayout.Var().Name()
+)
+
 // Backend extends the base backend interface with specific information about local backends.
 type Backend interface {
 	backend.Backend
@@ -91,29 +115,65 @@ type localBackend struct {
 	currentProject atomic.Pointer[workspace.Project]
 
 	// The store controls the layout of stacks in the backend.
+	// We use different layouts based on the version of the backend
+	// specified in the metadata file.
+	// If the metadata file is missing, we use the legacy layout.
 	store referenceStore
 }
 
 type localBackendReference struct {
-	name tokens.Name
+	name    tokens.Name
+	project tokens.Name
+
+	// A thread-safe way to get the current project.
+	// The function reference or the pointer returned by the function may be nil.
+	currentProject func() *workspace.Project
 
 	// referenceStore that created this reference.
+	//
+	// This is necessary because
+	// the referenceStore for a backend may change over time,
+	// but the store for this reference should not.
 	store referenceStore
 }
 
 func (r *localBackendReference) String() string {
-	return string(r.name)
+	// If project is blank this is a legacy non-project scoped stack reference, just return the name.
+	if r.project == "" {
+		return string(r.name)
+	}
+
+	if r.currentProject != nil {
+		proj := r.currentProject()
+		// For project scoped references when stringifying backend references,
+		// we take the current project (if present) into account.
+		// If the project names match, we can elide them.
+		if proj != nil && string(r.project) == string(proj.Name) {
+			return string(r.name)
+		}
+	}
+
+	// Else return a new style fully qualified reference.
+	return fmt.Sprintf("organization/%s/%s", r.project, r.name)
 }
 
 func (r *localBackendReference) Name() tokens.Name {
 	return r.name
 }
 
+func (r *localBackendReference) Project() tokens.Name {
+	return r.project
+}
+
 func (r *localBackendReference) FullyQualifiedName() tokens.QName {
-	return r.Name().Q()
+	if r.project == "" {
+		return r.name.Q()
+	}
+	return tokens.QName(fmt.Sprintf("organization/%s/%s", r.project, r.name))
 }
 
 // Helper methods that delegate to the underlying referenceStore.
+func (r *localBackendReference) Validate() error       { return r.store.ValidateReference(r) }
 func (r *localBackendReference) StackBasePath() string { return r.store.StackBasePath(r) }
 func (r *localBackendReference) HistoryDir() string    { return r.store.HistoryDir(r) }
 func (r *localBackendReference) BackupDir() string     { return r.store.BackupDir(r) }
@@ -129,6 +189,10 @@ func IsFileStateBackendURL(urlstr string) bool {
 
 const FilePathPrefix = "file://"
 
+// New constructs a new filestate backend,
+// using the given URL as the root for storage.
+// The URL must use one of the schemes supported by the go-cloud blob package.
+// Thes inclue: file, s3, gs, azblob.
 func New(ctx context.Context, d diag.Sink, originalURL string, project *workspace.Project) (Backend, error) {
 	if !IsFileStateBackendURL(originalURL) {
 		return nil, fmt.Errorf("local URL %s has an illegal prefix; expected one of: %s",
@@ -190,22 +254,66 @@ func New(ctx context.Context, d diag.Sink, originalURL string, project *workspac
 		bucket:      wbucket,
 		lockID:      lockID.String(),
 		gzip:        gzipCompression,
-		store:       newLegacyReferenceStore(wbucket),
 	}
 	backend.currentProject.Store(project)
 
 	// Read the Pulumi state metadata
 	// and ensure that it is compatible with this version of the CLI.
+	// The version in the metadata file informs which store we use.
 	meta, err := ensurePulumiMeta(ctx, wbucket)
 	if err != nil {
 		return nil, err
 	}
-	if meta.Version != 0 {
+
+	// projectMode tracks whether the current state supports project-scoped stacks.
+	// Historically, the filestate backend did not support this.
+	// To avoid breaking old stacks, we use legacy mode for existing states.
+	// We use project mode only if one of the following is true:
+	//
+	//  - The state has a single .pulumi/meta.yaml file
+	//    and the version is 1 or greater.
+	//  - The state is entirely new
+	//    so there's no risk of breaking old stacks.
+	//
+	// All actual logic of project mode vs legacy mode is handled by the referenceStore.
+	// This boolean just helps us warn users about unmigrated stacks.
+	var projectMode bool
+	switch meta.Version {
+	case 0:
+		backend.store = newLegacyReferenceStore(wbucket)
+	case 1:
+		backend.store = newProjectReferenceStore(wbucket, backend.currentProject.Load)
+		projectMode = true
+	default:
 		return nil, fmt.Errorf(
 			"state store unsupported: 'meta.yaml' version (%d) is not supported "+
 				"by this version of the Pulumi CLI", meta.Version)
 	}
 
+	// If we're not in project mode, or we've disabled the warning, we're done.
+	if !projectMode || cmdutil.IsTruthy(os.Getenv(PulumiFilestateNoLegacyWarningEnvVar)) {
+		return backend, nil
+	}
+	// Otherwise, warn about any old stack files.
+	// This is possible if a user creates a new stack with a new CLI,
+	// but someone else interacts with the same state with an old CLI.
+
+	refs, err := newLegacyReferenceStore(wbucket).ListReferences()
+	if err != nil {
+		// If there's an error listing don't fail, just don't print the warnings
+		return backend, nil
+	}
+	if len(refs) == 0 {
+		return backend, nil
+	}
+
+	var msg strings.Builder
+	msg.WriteString("Found legacy stack files in state store:\n")
+	for _, ref := range refs {
+		fmt.Fprintf(&msg, "  - %s\n", ref.Name())
+	}
+	msg.WriteString("Set PULUMI_SELF_MANAGED_STATE_NO_LEGACY_WARNING=1 to disable this warning.")
+	d.Warningf(diag.Message("", msg.String()))
 	return backend, nil
 }
 
@@ -267,7 +375,7 @@ func (b *localBackend) getReference(ref backend.StackReference) (*localBackendRe
 	if !ok {
 		return nil, fmt.Errorf("bad stack reference type")
 	}
-	return stackRef, nil
+	return stackRef, stackRef.Validate()
 }
 
 func (b *localBackend) local() {}
@@ -330,10 +438,56 @@ func (b *localBackend) ValidateStackName(stackRef string) error {
 }
 
 func (b *localBackend) DoesProjectExist(ctx context.Context, projectName string) (bool, error) {
-	// Local backends don't really have multiple projects, so just return false here.
+	projStore, ok := b.store.(*projectReferenceStore)
+	if !ok {
+		// Legacy stores don't have projects
+		// so the project does not exist.
+		return false, nil
+	}
+
+	// TODO[pulumi/pulumi#12547]:
+	// This could be faster if we list "$project/" instead of all projects.
+	projects, err := projStore.ListProjects()
+	if err != nil {
+		return false, err
+	}
+
+	for _, project := range projects {
+		if string(project) == projectName {
+			return true, nil
+		}
+	}
+
 	return false, nil
 }
 
+// Confirm the specified stack's project doesn't contradict the meta.yaml of the current project.
+// If the CWD is not in a Pulumi project, does not contradict.
+// If the project name in Pulumi.yaml is "foo", a stack with a name of bar/foo should not work.
+func currentProjectContradictsWorkspace(stack *localBackendReference) bool {
+	contract.Requiref(stack != nil, "stack", "is nil")
+
+	if stack.project == "" {
+		return false
+	}
+
+	projPath, err := workspace.DetectProjectPath()
+	if err != nil {
+		return false
+	}
+
+	if projPath == "" {
+		return false
+	}
+
+	proj, err := workspace.LoadProject(projPath)
+	if err != nil {
+		return false
+	}
+
+	return proj.Name.String() != stack.project.String()
+}
+
 func (b *localBackend) CreateStack(ctx context.Context, stackRef backend.StackReference,
 	root string, opts *backend.CreateStackOptions,
 ) (backend.Stack, error) {
@@ -352,6 +506,10 @@ func (b *localBackend) CreateStack(ctx context.Context, stackRef backend.StackRe
 	}
 	defer b.Unlock(ctx, stackRef)
 
+	if currentProjectContradictsWorkspace(localStackRef) {
+		return nil, fmt.Errorf("provided project name %q doesn't match Pulumi.yaml", localStackRef.project)
+	}
+
 	stackName := localStackRef.FullyQualifiedName()
 	if stackName == "" {
 		return nil, errors.New("invalid empty stack name")
@@ -617,6 +775,10 @@ func (b *localBackend) apply(
 		return nil, nil, result.FromError(err)
 	}
 
+	if currentProjectContradictsWorkspace(localStackRef) {
+		return nil, nil, result.Errorf("provided project name %q doesn't match Pulumi.yaml", localStackRef.project)
+	}
+
 	stackName := stackRef.FullyQualifiedName()
 	actionLabel := backend.ActionLabel(kind, opts.DryRun)
 
diff --git a/pkg/backend/filestate/backend_legacy_test.go b/pkg/backend/filestate/backend_legacy_test.go
new file mode 100644
index 000000000000..479be86159e1
--- /dev/null
+++ b/pkg/backend/filestate/backend_legacy_test.go
@@ -0,0 +1,374 @@
+package filestate
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"os"
+	"path"
+	"path/filepath"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/pulumi/pulumi/pkg/v3/backend"
+	"github.com/pulumi/pulumi/pkg/v3/resource/deploy"
+	"github.com/pulumi/pulumi/pkg/v3/resource/stack"
+	"github.com/pulumi/pulumi/pkg/v3/secrets/b64"
+	"github.com/pulumi/pulumi/sdk/v3/go/common/apitype"
+	"github.com/pulumi/pulumi/sdk/v3/go/common/encoding"
+	"github.com/pulumi/pulumi/sdk/v3/go/common/resource"
+	"github.com/pulumi/pulumi/sdk/v3/go/common/testing/diagtest"
+)
+
+// This file contains copies of old backend tests
+// that were upgraded to run with project support.
+// This duplicates those tests to run with legacy, non-project state,
+// validating that the legacy behavior is preserved.
+
+//nolint:paralleltest // mutates environment variables
+func TestListStacksWithMultiplePassphrases_legacy(t *testing.T) {
+	// Login to a temp dir filestate backend
+	tmpDir := markLegacyStore(t, t.TempDir())
+	ctx := context.Background()
+	b, err := New(ctx, diagtest.LogSink(t), "file://"+filepath.ToSlash(tmpDir), nil)
+	assert.NoError(t, err)
+
+	// Create stack "a" and import a checkpoint with a secret
+	aStackRef, err := b.ParseStackReference("a")
+	assert.NoError(t, err)
+	aStack, err := b.CreateStack(ctx, aStackRef, "", nil)
+	assert.NoError(t, err)
+	assert.NotNil(t, aStack)
+	defer func() {
+		t.Setenv("PULUMI_CONFIG_PASSPHRASE", "abc123")
+		_, err := b.RemoveStack(ctx, aStack, true)
+		assert.NoError(t, err)
+	}()
+	deployment, err := makeUntypedDeployment("a", "abc123",
+		"v1:4iF78gb0nF0=:v1:Co6IbTWYs/UdrjgY:FSrAWOFZnj9ealCUDdJL7LrUKXX9BA==")
+	assert.NoError(t, err)
+	t.Setenv("PULUMI_CONFIG_PASSPHRASE", "abc123")
+	err = b.ImportDeployment(ctx, aStack, deployment)
+	assert.NoError(t, err)
+
+	// Create stack "b" and import a checkpoint with a secret
+	bStackRef, err := b.ParseStackReference("b")
+	assert.NoError(t, err)
+	bStack, err := b.CreateStack(ctx, bStackRef, "", nil)
+	assert.NoError(t, err)
+	assert.NotNil(t, bStack)
+	defer func() {
+		t.Setenv("PULUMI_CONFIG_PASSPHRASE", "123abc")
+		_, err := b.RemoveStack(ctx, bStack, true)
+		assert.NoError(t, err)
+	}()
+	deployment, err = makeUntypedDeployment("b", "123abc",
+		"v1:C7H2a7/Ietk=:v1:yfAd1zOi6iY9DRIB:dumdsr+H89VpHIQWdB01XEFqYaYjAg==")
+	assert.NoError(t, err)
+	t.Setenv("PULUMI_CONFIG_PASSPHRASE", "123abc")
+	err = b.ImportDeployment(ctx, bStack, deployment)
+	assert.NoError(t, err)
+
+	// Remove the config passphrase so that we can no longer deserialize the checkpoints
+	err = os.Unsetenv("PULUMI_CONFIG_PASSPHRASE")
+	assert.NoError(t, err)
+
+	// Ensure that we can list the stacks we created even without a passphrase
+	stacks, outContToken, err := b.ListStacks(ctx, backend.ListStacksFilter{}, nil /* inContToken */)
+	assert.NoError(t, err)
+	assert.Nil(t, outContToken)
+	assert.Len(t, stacks, 2)
+	for _, stack := range stacks {
+		assert.NotNil(t, stack.ResourceCount())
+		assert.Equal(t, 1, *stack.ResourceCount())
+	}
+}
+
+func TestDrillError_legacy(t *testing.T) {
+	t.Parallel()
+
+	// Login to a temp dir filestate backend
+	tmpDir := markLegacyStore(t, t.TempDir())
+	ctx := context.Background()
+	b, err := New(ctx, diagtest.LogSink(t), "file://"+filepath.ToSlash(tmpDir), nil)
+	assert.NoError(t, err)
+
+	// Get a non-existent stack and expect a nil error because it won't be found.
+	stackRef, err := b.ParseStackReference("dev")
+	if err != nil {
+		t.Fatalf("unexpected error %v when parsing stack reference", err)
+	}
+	_, err = b.GetStack(ctx, stackRef)
+	assert.Nil(t, err)
+}
+
+func TestCancel_legacy(t *testing.T) {
+	t.Parallel()
+
+	// Login to a temp dir filestate backend
+	tmpDir := markLegacyStore(t, t.TempDir())
+	ctx := context.Background()
+	b, err := New(ctx, diagtest.LogSink(t), "file://"+filepath.ToSlash(tmpDir), nil)
+	assert.NoError(t, err)
+
+	// Check that trying to cancel a stack that isn't created yet doesn't error
+	aStackRef, err := b.ParseStackReference("a")
+	assert.NoError(t, err)
+	err = b.CancelCurrentUpdate(ctx, aStackRef)
+	assert.NoError(t, err)
+
+	// Check that trying to cancel a stack that isn't locked doesn't error
+	aStack, err := b.CreateStack(ctx, aStackRef, "", nil)
+	assert.NoError(t, err)
+	assert.NotNil(t, aStack)
+	err = b.CancelCurrentUpdate(ctx, aStackRef)
+	assert.NoError(t, err)
+
+	// Locking and lock checks are only part of the internal interface
+	lb, ok := b.(*localBackend)
+	assert.True(t, ok)
+	assert.NotNil(t, lb)
+
+	// Lock the stack and check CancelCurrentUpdate deletes the lock file
+	err = lb.Lock(ctx, aStackRef)
+	assert.NoError(t, err)
+	// check the lock file exists
+	lockExists, err := lb.bucket.Exists(ctx, lb.lockPath(aStackRef))
+	assert.NoError(t, err)
+	assert.True(t, lockExists)
+	// Call CancelCurrentUpdate
+	err = lb.CancelCurrentUpdate(ctx, aStackRef)
+	assert.NoError(t, err)
+	// Now check the lock file no longer exists
+	lockExists, err = lb.bucket.Exists(ctx, lb.lockPath(aStackRef))
+	assert.NoError(t, err)
+	assert.False(t, lockExists)
+
+	// Make another filestate backend which will have a different lockId
+	ob, err := New(ctx, diagtest.LogSink(t), "file://"+filepath.ToSlash(tmpDir), nil)
+	assert.NoError(t, err)
+	otherBackend, ok := ob.(*localBackend)
+	assert.True(t, ok)
+	assert.NotNil(t, lb)
+
+	// Lock the stack with this new backend, then check that checkForLocks on the first backend now errors
+	err = otherBackend.Lock(ctx, aStackRef)
+	assert.NoError(t, err)
+	err = lb.checkForLock(ctx, aStackRef)
+	assert.Error(t, err)
+	// Now call CancelCurrentUpdate and check that checkForLocks no longer errors
+	err = lb.CancelCurrentUpdate(ctx, aStackRef)
+	assert.NoError(t, err)
+	err = lb.checkForLock(ctx, aStackRef)
+	assert.NoError(t, err)
+}
+
+func TestRemoveMakesBackups_legacy(t *testing.T) {
+	t.Parallel()
+
+	// Login to a temp dir filestate backend
+	tmpDir := markLegacyStore(t, t.TempDir())
+	ctx := context.Background()
+	b, err := New(ctx, diagtest.LogSink(t), "file://"+filepath.ToSlash(tmpDir), nil)
+	assert.NoError(t, err)
+
+	// Grab the bucket interface to test with
+	lb, ok := b.(*localBackend)
+	assert.True(t, ok)
+	assert.NotNil(t, lb)
+
+	// Check that creating a new stack doesn't make a backup file
+	aStackRef, err := lb.parseStackReference("a")
+	assert.NoError(t, err)
+	aStack, err := b.CreateStack(ctx, aStackRef, "", nil)
+	assert.NoError(t, err)
+	assert.NotNil(t, aStack)
+
+	// Check the stack file now exists, but the backup file doesn't
+	stackFileExists, err := lb.bucket.Exists(ctx, lb.stackPath(aStackRef))
+	assert.NoError(t, err)
+	assert.True(t, stackFileExists)
+	backupFileExists, err := lb.bucket.Exists(ctx, lb.stackPath(aStackRef)+".bak")
+	assert.NoError(t, err)
+	assert.False(t, backupFileExists)
+
+	// Now remove the stack
+	removed, err := b.RemoveStack(ctx, aStack, false)
+	assert.NoError(t, err)
+	assert.False(t, removed)
+
+	// Check the stack file is now gone, but the backup file exists
+	stackFileExists, err = lb.bucket.Exists(ctx, lb.stackPath(aStackRef))
+	assert.NoError(t, err)
+	assert.False(t, stackFileExists)
+	backupFileExists, err = lb.bucket.Exists(ctx, lb.stackPath(aStackRef)+".bak")
+	assert.NoError(t, err)
+	assert.True(t, backupFileExists)
+}
+
+func TestRenameWorks_legacy(t *testing.T) {
+	t.Parallel()
+
+	// Login to a temp dir filestate backend
+	tmpDir := markLegacyStore(t, t.TempDir())
+	ctx := context.Background()
+	b, err := New(ctx, diagtest.LogSink(t), "file://"+filepath.ToSlash(tmpDir), nil)
+	assert.NoError(t, err)
+
+	// Grab the bucket interface to test with
+	lb, ok := b.(*localBackend)
+	assert.True(t, ok)
+	assert.NotNil(t, lb)
+
+	// Create a new stack
+	aStackRef, err := lb.parseStackReference("a")
+	assert.NoError(t, err)
+	aStack, err := b.CreateStack(ctx, aStackRef, "", nil)
+	assert.NoError(t, err)
+	assert.NotNil(t, aStack)
+
+	// Check the stack file now exists
+	stackFileExists, err := lb.bucket.Exists(ctx, lb.stackPath(aStackRef))
+	assert.NoError(t, err)
+	assert.True(t, stackFileExists)
+
+	// Fake up some history
+	err = lb.addToHistory(aStackRef, backend.UpdateInfo{Kind: apitype.DestroyUpdate})
+	assert.NoError(t, err)
+	// And pollute the history folder
+	err = lb.bucket.WriteAll(ctx, path.Join(aStackRef.HistoryDir(), "randomfile.txt"), []byte{0, 13}, nil)
+	assert.NoError(t, err)
+
+	// Rename the stack
+	bStackRefI, err := b.RenameStack(ctx, aStack, "b")
+	assert.NoError(t, err)
+	assert.Equal(t, "b", bStackRefI.String())
+	bStackRef := bStackRefI.(*localBackendReference)
+
+	// Check the new stack file now exists and the old one is gone
+	stackFileExists, err = lb.bucket.Exists(ctx, lb.stackPath(bStackRef))
+	assert.NoError(t, err)
+	assert.True(t, stackFileExists)
+	stackFileExists, err = lb.bucket.Exists(ctx, lb.stackPath(aStackRef))
+	assert.NoError(t, err)
+	assert.False(t, stackFileExists)
+
+	// Rename again
+	bStack, err := b.GetStack(ctx, bStackRef)
+	assert.NoError(t, err)
+	cStackRefI, err := b.RenameStack(ctx, bStack, "c")
+	assert.NoError(t, err)
+	assert.Equal(t, "c", cStackRefI.String())
+	cStackRef := cStackRefI.(*localBackendReference)
+
+	// Check the new stack file now exists and the old one is gone
+	stackFileExists, err = lb.bucket.Exists(ctx, lb.stackPath(cStackRef))
+	assert.NoError(t, err)
+	assert.True(t, stackFileExists)
+	stackFileExists, err = lb.bucket.Exists(ctx, lb.stackPath(bStackRef))
+	assert.NoError(t, err)
+	assert.False(t, stackFileExists)
+
+	// Check we can still get the history
+	history, err := b.GetHistory(ctx, cStackRef, 10, 0)
+	assert.NoError(t, err)
+	assert.Len(t, history, 1)
+	assert.Equal(t, apitype.DestroyUpdate, history[0].Kind)
+}
+
+// Regression test for https://github.com/pulumi/pulumi/issues/10439
+func TestHtmlEscaping_legacy(t *testing.T) {
+	t.Parallel()
+
+	sm := b64.NewBase64SecretsManager()
+	resources := []*resource.State{
+		{
+			URN:  resource.NewURN("a", "proj", "d:e:f", "a:b:c", "name"),
+			Type: "a:b:c",
+			Inputs: resource.PropertyMap{
+				resource.PropertyKey("html"): resource.NewStringProperty("<html@tags>"),
+			},
+		},
+	}
+
+	snap := deploy.NewSnapshot(deploy.Manifest{}, sm, resources, nil)
+
+	sdep, err := stack.SerializeDeployment(snap, snap.SecretsManager, false /* showSecrsts */)
+	assert.NoError(t, err)
+
+	data, err := encoding.JSON.Marshal(sdep)
+	assert.NoError(t, err)
+
+	// Ensure data has the string contents "<html@tags>"", not "\u003chtml\u0026tags\u003e"
+	// ImportDeployment below should not modify the data
+	assert.Contains(t, string(data), "<html@tags>")
+
+	udep := &apitype.UntypedDeployment{
+		Version:    3,
+		Deployment: json.RawMessage(data),
+	}
+
+	// Login to a temp dir filestate backend
+	tmpDir := markLegacyStore(t, t.TempDir())
+	ctx := context.Background()
+	b, err := New(ctx, diagtest.LogSink(t), "file://"+filepath.ToSlash(tmpDir), nil)
+	assert.NoError(t, err)
+
+	// Create stack "a" and import a checkpoint with a secret
+	aStackRef, err := b.ParseStackReference("a")
+	assert.NoError(t, err)
+	aStack, err := b.CreateStack(ctx, aStackRef, "", nil)
+	assert.NoError(t, err)
+	assert.NotNil(t, aStack)
+	err = b.ImportDeployment(ctx, aStack, udep)
+	assert.NoError(t, err)
+
+	// Ensure the file has the string contents "<html@tags>"", not "\u003chtml\u0026tags\u003e"
+
+	// Grab the bucket interface to read the file with
+	lb, ok := b.(*localBackend)
+	assert.True(t, ok)
+	assert.NotNil(t, lb)
+
+	chkpath := lb.stackPath(aStackRef.(*localBackendReference))
+	bytes, err := lb.bucket.ReadAll(context.Background(), chkpath)
+	assert.NoError(t, err)
+	state := string(bytes)
+	assert.Contains(t, state, "<html@tags>")
+}
+
+func TestLocalBackendRejectsStackInitOptions_legacy(t *testing.T) {
+	t.Parallel()
+
+	// Here, we provide options that illegally specify a team on a
+	// backend that does not support teams. We expect this to create
+	// an error later when we call CreateStack.
+	illegalOptions := &backend.CreateStackOptions{Teams: []string{"red-team"}}
+
+	// • Create a mock local backend
+	tmpDir := markLegacyStore(t, t.TempDir())
+	dirURI := fmt.Sprintf("file://%s", filepath.ToSlash(tmpDir))
+	local, err := New(context.Background(), diagtest.LogSink(t), dirURI, nil)
+	assert.NoError(t, err)
+	ctx := context.Background()
+
+	// • Simulate `pulumi stack init`, passing non-nil init options
+	fakeStackRef, err := local.ParseStackReference("foobar")
+	assert.NoError(t, err)
+	_, err = local.CreateStack(ctx, fakeStackRef, "", illegalOptions)
+	assert.ErrorIs(t, err, backend.ErrTeamsNotSupported)
+}
+
+// markLegacyStore marks the given directory as a legacy store.
+// This is done by dropping a single file into the bookkeeping directory.
+// ensurePulumiMeta will treat this as a legacy store if the directory exists.
+//
+// Returns the directory that was marked.
+func markLegacyStore(t *testing.T, dir string) string {
+	metaPath := filepath.Join(dir, pulumiMetaPath)
+	require.NoError(t, os.MkdirAll(filepath.Dir(metaPath), 0o755))
+	require.NoError(t, os.WriteFile(metaPath, []byte(`version: 0`), 0o600))
+	return dir
+}
diff --git a/pkg/backend/filestate/backend_test.go b/pkg/backend/filestate/backend_test.go
index a15e226f2451..c9448543bf56 100644
--- a/pkg/backend/filestate/backend_test.go
+++ b/pkg/backend/filestate/backend_test.go
@@ -1,13 +1,16 @@
 package filestate
 
 import (
+	"bytes"
 	"context"
 	"encoding/json"
 	"fmt"
+	"io"
 	"os"
 	"path"
 	"path/filepath"
 	"runtime"
+	"sync"
 	"testing"
 	"time"
 
@@ -23,11 +26,14 @@ import (
 	"github.com/pulumi/pulumi/pkg/v3/secrets/b64"
 	"github.com/pulumi/pulumi/pkg/v3/secrets/passphrase"
 	"github.com/pulumi/pulumi/sdk/v3/go/common/apitype"
+	"github.com/pulumi/pulumi/sdk/v3/go/common/diag"
+	"github.com/pulumi/pulumi/sdk/v3/go/common/diag/colors"
 	"github.com/pulumi/pulumi/sdk/v3/go/common/encoding"
 	"github.com/pulumi/pulumi/sdk/v3/go/common/resource"
 	"github.com/pulumi/pulumi/sdk/v3/go/common/resource/config"
 	"github.com/pulumi/pulumi/sdk/v3/go/common/testing/diagtest"
 	"github.com/pulumi/pulumi/sdk/v3/go/common/tokens"
+	"github.com/pulumi/pulumi/sdk/v3/go/common/workspace"
 )
 
 func TestMassageBlobPath(t *testing.T) {
@@ -163,7 +169,7 @@ func TestListStacksWithMultiplePassphrases(t *testing.T) {
 	assert.NoError(t, err)
 
 	// Create stack "a" and import a checkpoint with a secret
-	aStackRef, err := b.ParseStackReference("a")
+	aStackRef, err := b.ParseStackReference("organization/project/a")
 	assert.NoError(t, err)
 	aStack, err := b.CreateStack(ctx, aStackRef, "", nil)
 	assert.NoError(t, err)
@@ -181,7 +187,7 @@ func TestListStacksWithMultiplePassphrases(t *testing.T) {
 	assert.NoError(t, err)
 
 	// Create stack "b" and import a checkpoint with a secret
-	bStackRef, err := b.ParseStackReference("b")
+	bStackRef, err := b.ParseStackReference("organization/project/b")
 	assert.NoError(t, err)
 	bStack, err := b.CreateStack(ctx, bStackRef, "", nil)
 	assert.NoError(t, err)
@@ -223,7 +229,7 @@ func TestDrillError(t *testing.T) {
 	assert.NoError(t, err)
 
 	// Get a non-existent stack and expect a nil error because it won't be found.
-	stackRef, err := b.ParseStackReference("dev")
+	stackRef, err := b.ParseStackReference("organization/project/dev")
 	if err != nil {
 		t.Fatalf("unexpected error %v when parsing stack reference", err)
 	}
@@ -241,7 +247,7 @@ func TestCancel(t *testing.T) {
 	assert.NoError(t, err)
 
 	// Check that trying to cancel a stack that isn't created yet doesn't error
-	aStackRef, err := b.ParseStackReference("a")
+	aStackRef, err := b.ParseStackReference("organization/project/a")
 	assert.NoError(t, err)
 	err = b.CancelCurrentUpdate(ctx, aStackRef)
 	assert.NoError(t, err)
@@ -307,7 +313,7 @@ func TestRemoveMakesBackups(t *testing.T) {
 	assert.NotNil(t, lb)
 
 	// Check that creating a new stack doesn't make a backup file
-	aStackRef, err := lb.parseStackReference("a")
+	aStackRef, err := lb.parseStackReference("organization/project/a")
 	assert.NoError(t, err)
 	aStack, err := b.CreateStack(ctx, aStackRef, "", nil)
 	assert.NoError(t, err)
@@ -350,7 +356,7 @@ func TestRenameWorks(t *testing.T) {
 	assert.NotNil(t, lb)
 
 	// Create a new stack
-	aStackRef, err := lb.parseStackReference("a")
+	aStackRef, err := lb.parseStackReference("organization/project/a")
 	assert.NoError(t, err)
 	aStack, err := b.CreateStack(ctx, aStackRef, "", nil)
 	assert.NoError(t, err)
@@ -369,9 +375,9 @@ func TestRenameWorks(t *testing.T) {
 	assert.NoError(t, err)
 
 	// Rename the stack
-	bStackRefI, err := b.RenameStack(ctx, aStack, "b")
+	bStackRefI, err := b.RenameStack(ctx, aStack, "organization/project/b")
 	assert.NoError(t, err)
-	assert.Equal(t, "b", bStackRefI.String())
+	assert.Equal(t, "organization/project/b", bStackRefI.String())
 	bStackRef := bStackRefI.(*localBackendReference)
 
 	// Check the new stack file now exists and the old one is gone
@@ -385,9 +391,9 @@ func TestRenameWorks(t *testing.T) {
 	// Rename again
 	bStack, err := b.GetStack(ctx, bStackRef)
 	assert.NoError(t, err)
-	cStackRefI, err := b.RenameStack(ctx, bStack, "c")
+	cStackRefI, err := b.RenameStack(ctx, bStack, "organization/project/c")
 	assert.NoError(t, err)
-	assert.Equal(t, "c", cStackRefI.String())
+	assert.Equal(t, "organization/project/c", cStackRefI.String())
 	cStackRef := cStackRefI.(*localBackendReference)
 
 	// Check the new stack file now exists and the old one is gone
@@ -419,7 +425,6 @@ func TestLoginToNonExistingFolderFails(t *testing.T) {
 // an error when the stack name is the empty string.TestParseEmptyStackFails
 func TestParseEmptyStackFails(t *testing.T) {
 	t.Parallel()
-
 	tmpDir := t.TempDir()
 	ctx := context.Background()
 	b, err := New(ctx, diagtest.LogSink(t), "file://"+filepath.ToSlash(tmpDir), nil)
@@ -468,7 +473,7 @@ func TestHtmlEscaping(t *testing.T) {
 	assert.NoError(t, err)
 
 	// Create stack "a" and import a checkpoint with a secret
-	aStackRef, err := b.ParseStackReference("a")
+	aStackRef, err := b.ParseStackReference("organization/project/a")
 	assert.NoError(t, err)
 	aStack, err := b.CreateStack(ctx, aStackRef, "", nil)
 	assert.NoError(t, err)
@@ -505,12 +510,321 @@ func TestLocalBackendRejectsStackInitOptions(t *testing.T) {
 	ctx := context.Background()
 
 	// • Simulate `pulumi stack init`, passing non-nil init options
-	fakeStackRef, err := local.ParseStackReference("foobar")
+	fakeStackRef, err := local.ParseStackReference("organization/b/foobar")
 	assert.NoError(t, err)
 	_, err = local.CreateStack(ctx, fakeStackRef, "", illegalOptions)
 	assert.ErrorIs(t, err, backend.ErrTeamsNotSupported)
 }
 
+func TestLegacyFolderStructure(t *testing.T) {
+	t.Parallel()
+
+	// Make a dummy stack file in the legacy location
+	tmpDir := t.TempDir()
+	err := os.MkdirAll(path.Join(tmpDir, ".pulumi", "stacks"), os.ModePerm)
+	require.NoError(t, err)
+	err = os.WriteFile(path.Join(tmpDir, ".pulumi", "stacks", "a.json"), []byte("{}"), os.ModePerm)
+	require.NoError(t, err)
+
+	// Login to a temp dir filestate backend
+	ctx := context.Background()
+	b, err := New(ctx, diagtest.LogSink(t), "file://"+filepath.ToSlash(tmpDir), nil)
+	require.NoError(t, err)
+	// Check the backend says it's NOT in project mode
+	lb, ok := b.(*localBackend)
+	assert.True(t, ok)
+	assert.NotNil(t, lb)
+	assert.IsType(t, &legacyReferenceStore{}, lb.store)
+
+	// Check that list stack shows that stack
+	stacks, token, err := b.ListStacks(ctx, backend.ListStacksFilter{}, nil /* inContToken */)
+	assert.NoError(t, err)
+	assert.Nil(t, token)
+	assert.Len(t, stacks, 1)
+	assert.Equal(t, "a", stacks[0].Name().String())
+
+	// Create a new non-project stack
+	bRef, err := b.ParseStackReference("b")
+	assert.NoError(t, err)
+	assert.Equal(t, "b", bRef.String())
+	bStack, err := b.CreateStack(ctx, bRef, "", nil)
+	assert.NoError(t, err)
+	assert.Equal(t, "b", bStack.Ref().String())
+	assert.FileExists(t, path.Join(tmpDir, ".pulumi", "stacks", "b.json"))
+}
+
+//nolint:paralleltest // uses t.Setenv
+func TestOptIntoLegacyFolderStructure(t *testing.T) {
+	t.Setenv("PULUMI_SELF_MANAGED_STATE_LEGACY_LAYOUT", "true")
+
+	tmpDir := t.TempDir()
+	ctx := context.Background()
+	b, err := New(ctx, diagtest.LogSink(t), "file://"+filepath.ToSlash(tmpDir), nil)
+	require.NoError(t, err)
+
+	// Verify that a new stack is created in the legacy location.
+	foo, err := b.ParseStackReference("foo")
+	require.NoError(t, err)
+
+	_, err = b.CreateStack(ctx, foo, "", nil)
+	require.NoError(t, err)
+	assert.FileExists(t, filepath.Join(tmpDir, ".pulumi", "stacks", "foo.json"))
+}
+
+// Verifies that the StackReference.String method
+// takes the current project name into account,
+// even if the current project name changes
+// after the stack reference is created.
+func TestStackReferenceString_currentProjectChange(t *testing.T) {
+	t.Parallel()
+
+	dir := t.TempDir()
+	ctx := context.Background()
+
+	b, err := New(ctx, diagtest.LogSink(t), "file://"+filepath.ToSlash(dir), nil)
+	require.NoError(t, err)
+
+	foo, err := b.ParseStackReference("organization/proj1/foo")
+	require.NoError(t, err)
+
+	bar, err := b.ParseStackReference("organization/proj2/bar")
+	require.NoError(t, err)
+
+	assert.Equal(t, "organization/proj1/foo", foo.String())
+	assert.Equal(t, "organization/proj2/bar", bar.String())
+
+	// Change the current project name
+	b.SetCurrentProject(&workspace.Project{Name: "proj1"})
+
+	assert.Equal(t, "foo", foo.String())
+	assert.Equal(t, "organization/proj2/bar", bar.String())
+}
+
+// Verifies that there's no data race in calling StackReference.String
+// and localBackend.SetCurrentProject concurrently.
+func TestStackReferenceString_currentProjectChange_race(t *testing.T) {
+	t.Parallel()
+
+	const N = 1000
+
+	dir := t.TempDir()
+	ctx := context.Background()
+
+	b, err := New(ctx, diagtest.LogSink(t), "file://"+filepath.ToSlash(dir), nil)
+	require.NoError(t, err)
+
+	projects := make([]*workspace.Project, N)
+	refs := make([]backend.StackReference, N)
+	for i := 0; i < N; i++ {
+		name := fmt.Sprintf("proj%d", i)
+		projects[i] = &workspace.Project{Name: tokens.PackageName(name)}
+		refs[i], err = b.ParseStackReference(fmt.Sprintf("organization/%v/foo", name))
+		require.NoError(t, err)
+	}
+
+	// To exercise this data race, we'll have two goroutines.
+	// One goroutine will call StackReference.String repeatedly
+	// on all the stack references,
+	// and the other goroutine will call localBackend.SetCurrentProject
+	// with all the projects.
+
+	var wg sync.WaitGroup
+	ready := make(chan struct{}) // both goroutines wait on this
+
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		<-ready
+		for i := 0; i < N; i++ {
+			_ = refs[i].String()
+		}
+	}()
+
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		<-ready
+		for i := 0; i < N; i++ {
+			b.SetCurrentProject(projects[i])
+		}
+	}()
+
+	close(ready) // start racing
+	wg.Wait()
+}
+
+func TestProjectFolderStructure(t *testing.T) {
+	t.Parallel()
+
+	// Login to a temp dir filestate backend
+	tmpDir := t.TempDir()
+	ctx := context.Background()
+	b, err := New(ctx, diagtest.LogSink(t), "file://"+filepath.ToSlash(tmpDir), nil)
+	assert.NoError(t, err)
+
+	// Check the backend says it's in project mode
+	lb, ok := b.(*localBackend)
+	assert.True(t, ok)
+	assert.NotNil(t, lb)
+	assert.IsType(t, &projectReferenceStore{}, lb.store)
+
+	// Make a dummy stack file in the new project location
+	err = os.MkdirAll(path.Join(tmpDir, ".pulumi", "stacks", "testproj"), os.ModePerm)
+	assert.NoError(t, err)
+	err = os.WriteFile(path.Join(tmpDir, ".pulumi", "stacks", "testproj", "a.json"), []byte("{}"), os.ModePerm)
+	assert.NoError(t, err)
+
+	// Check that testproj is reported as existing
+	exists, err := b.DoesProjectExist(ctx, "testproj")
+	assert.NoError(t, err)
+	assert.True(t, exists)
+
+	// Check that list stack shows that stack
+	stacks, token, err := b.ListStacks(ctx, backend.ListStacksFilter{}, nil /* inContToken */)
+	assert.NoError(t, err)
+	assert.Nil(t, token)
+	assert.Len(t, stacks, 1)
+	assert.Equal(t, "organization/testproj/a", stacks[0].Name().String())
+
+	// Create a new project stack
+	bRef, err := b.ParseStackReference("organization/testproj/b")
+	assert.NoError(t, err)
+	assert.Equal(t, "organization/testproj/b", bRef.String())
+	bStack, err := b.CreateStack(ctx, bRef, "", nil)
+	assert.NoError(t, err)
+	assert.Equal(t, "organization/testproj/b", bStack.Ref().String())
+	assert.FileExists(t, path.Join(tmpDir, ".pulumi", "stacks", "testproj", "b.json"))
+}
+
+func chdir(t *testing.T, dir string) {
+	cwd, err := os.Getwd()
+	require.NoError(t, err)
+	require.NoError(t, os.Chdir(dir)) // Set directory
+	t.Cleanup(func() {
+		require.NoError(t, os.Chdir(cwd)) // Restore directory
+		restoredDir, err := os.Getwd()
+		require.NoError(t, err)
+		require.Equal(t, cwd, restoredDir)
+	})
+}
+
+//nolint:paralleltest // mutates cwd
+func TestProjectNameMustMatch(t *testing.T) {
+	// Create a new project
+	projectDir := t.TempDir()
+	pyaml := filepath.Join(projectDir, "Pulumi.yaml")
+	err := os.WriteFile(pyaml, []byte("name: my-project\nruntime: test"), 0o600)
+	require.NoError(t, err)
+	proj, err := workspace.LoadProject(pyaml)
+	require.NoError(t, err)
+
+	chdir(t, projectDir)
+
+	// Login to a temp dir filestate backend
+	tmpDir := t.TempDir()
+	ctx := context.Background()
+	b, err := New(ctx, diagtest.LogSink(t), "file://"+filepath.ToSlash(tmpDir), proj)
+	require.NoError(t, err)
+
+	// Create a new implicit-project stack
+	aRef, err := b.ParseStackReference("a")
+	assert.NoError(t, err)
+	assert.Equal(t, "a", aRef.String())
+	aStack, err := b.CreateStack(ctx, aRef, "", nil)
+	assert.NoError(t, err)
+	assert.Equal(t, "a", aStack.Ref().String())
+	assert.FileExists(t, path.Join(tmpDir, ".pulumi", "stacks", "my-project", "a.json"))
+
+	// Create a new project stack with the wrong project name
+	bRef, err := b.ParseStackReference("organization/not-my-project/b")
+	assert.NoError(t, err)
+	assert.Equal(t, "organization/not-my-project/b", bRef.String())
+	bStack, err := b.CreateStack(ctx, bRef, "", nil)
+	assert.Error(t, err)
+	assert.Nil(t, bStack)
+
+	// Create a new project stack with the right project name
+	cRef, err := b.ParseStackReference("organization/my-project/c")
+	assert.NoError(t, err)
+	assert.Equal(t, "c", cRef.String())
+	cStack, err := b.CreateStack(ctx, cRef, "", nil)
+	assert.NoError(t, err)
+	assert.Equal(t, "c", cStack.Ref().String())
+	assert.FileExists(t, path.Join(tmpDir, ".pulumi", "stacks", "my-project", "c.json"))
+}
+
+//nolint:paralleltest // uses t.Setenv
+func TestNew_legacyFileWarning(t *testing.T) {
+	// Verifies the names of files printed in warnings
+	// when legacy files are found while running in project mode.
+
+	tests := []struct {
+		desc    string
+		files   map[string]string
+		env     map[string]string
+		wantOut string
+	}{
+		{
+			desc: "no legacy stacks",
+			files: map[string]string{
+				// Should ignore non-stack files.
+				".pulumi/foo/extraneous_file": "",
+			},
+		},
+		{
+			desc: "legacy stacks",
+			files: map[string]string{
+				".pulumi/stacks/a.json":     "{}",
+				".pulumi/stacks/b.json":     "{}",
+				".pulumi/stacks/c.json.bak": "{}", // should ignore backup files
+			},
+			wantOut: "warning: Found legacy stack files in state store:\n" +
+				"  - a\n" +
+				"  - b\n" +
+				"Set PULUMI_SELF_MANAGED_STATE_NO_LEGACY_WARNING=1 to disable this warning.\n",
+		},
+		{
+			desc: "warning opt-out",
+			files: map[string]string{
+				".pulumi/stacks/a.json": "{}",
+				".pulumi/stacks/b.json": "{}",
+			},
+			env: map[string]string{
+				"PULUMI_SELF_MANAGED_STATE_NO_LEGACY_WARNING": "true",
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.desc, func(t *testing.T) {
+			for k, v := range tt.env {
+				t.Setenv(k, v)
+			}
+
+			stateDir := t.TempDir()
+			bucket, err := fileblob.OpenBucket(stateDir, nil)
+			require.NoError(t, err)
+
+			ctx := context.Background()
+			require.NoError(t,
+				bucket.WriteAll(ctx, ".pulumi/meta.yaml", []byte("version: 1"), nil),
+				"write meta.yaml")
+
+			for path, contents := range tt.files {
+				require.NoError(t, bucket.WriteAll(ctx, path, []byte(contents), nil),
+					"write %q", path)
+			}
+
+			var buff bytes.Buffer
+			sink := diag.DefaultSink(io.Discard, &buff, diag.FormatOptions{Color: colors.Never})
+			_, err = New(ctx, sink, "file://"+filepath.ToSlash(stateDir), nil)
+			require.NoError(t, err)
+
+			assert.Equal(t, tt.wantOut, buff.String())
+		})
+	}
+}
+
 func TestNew_unsupportedStoreVersion(t *testing.T) {
 	t.Parallel()
 
diff --git a/pkg/backend/filestate/bucket.go b/pkg/backend/filestate/bucket.go
index d33303bdb70a..432bbdb9d650 100644
--- a/pkg/backend/filestate/bucket.go
+++ b/pkg/backend/filestate/bucket.go
@@ -6,6 +6,7 @@ import (
 	"io"
 	"path"
 	"path/filepath"
+	"strings"
 
 	"github.com/pulumi/pulumi/sdk/v3/go/common/util/logging"
 	"gocloud.dev/blob"
@@ -87,7 +88,9 @@ func listBucket(bucket Bucket, dir string) ([]*blob.ListObject, error) {
 
 // objectName returns the filename of a ListObject (an object from a bucket).
 func objectName(obj *blob.ListObject) string {
-	_, filename := path.Split(obj.Key)
+	// If obj.Key ends in "/" we want to trim that to get the name just before
+	key := strings.TrimSuffix(obj.Key, "/")
+	_, filename := path.Split(key)
 	return filename
 }
 
diff --git a/pkg/backend/filestate/meta.go b/pkg/backend/filestate/meta.go
index 38125f825244..eb377033c263 100644
--- a/pkg/backend/filestate/meta.go
+++ b/pkg/backend/filestate/meta.go
@@ -17,7 +17,9 @@ package filestate
 import (
 	"context"
 	"fmt"
+	"os"
 	"path/filepath"
+	"strconv"
 
 	"github.com/pulumi/pulumi/sdk/v3/go/common/util/contract"
 	"github.com/pulumi/pulumi/sdk/v3/go/common/workspace"
@@ -43,6 +45,7 @@ type pulumiMeta struct {
 	//
 	// Version 0 is the starting version.
 	// It does not support project-scoped stacks.
+	// Version 1 adds support for project-scoped stacks.
 	//
 	// Does not use "omitempty" to differentiate
 	// between a missing field and a zero value.
@@ -56,6 +59,10 @@ type pulumiMeta struct {
 //
 // If the bucket is empty, this will create a new metadata file
 // with the latest version number.
+// This can be overridden by setting the environment variable
+// "PULUMI_SELF_MANAGED_STATE_LEGACY_LAYOUT" to "1".
+// ensurePulumiMeta uses the provided 'getenv' function
+// to read the environment variable.
 func ensurePulumiMeta(ctx context.Context, b Bucket) (*pulumiMeta, error) {
 	meta, err := readPulumiMeta(ctx, b)
 	if err != nil {
@@ -66,13 +73,40 @@ func ensurePulumiMeta(ctx context.Context, b Bucket) (*pulumiMeta, error) {
 		return meta, nil
 	}
 
-	// If there's no metadata file, we need to create one
-	// with the latest version.
+	// If there's no metadata file, we need to create one.
+	// The version we pick for the new file decides how we lay out the state.
 	//
+	// - Version 0 is legacy mode, which is the old layout.
+	//   To avoid breaking old stacks, we want to use version 0
+	//   if the bucket is not empty.
+	//
+	// - Version 1 added support for project-scoped stacks.
+	//   For entirely new buckets, we'll use version 1
+	//   to give new users access to the latest features.
+	empty, err := isPulumiDirEmpty(ctx, b)
+	if err != nil {
+		return nil, err
+	}
+
+	useLegacy := !empty
+	if empty {
+		// Allow opting into legacy mode for new states
+		// by setting the environment variable.
+		v, err := strconv.ParseBool(os.Getenv(PulumiFilestateLegacyLayoutEnvVar))
+		if err == nil {
+			useLegacy = v
+		}
+	}
+
+	if useLegacy {
+		meta = &pulumiMeta{Version: 0}
+	} else {
+		meta = &pulumiMeta{Version: 1}
+	}
+
 	// Implementation detail:
-	// For version 0, we don't write the file.
-	// However, for future versions, we will write it.
-	meta = &pulumiMeta{Version: 0}
+	// For version 0, WriteTo won't write the metadata file.
+	// See [pulumiMeta.WriteTo] for details on why.
 	if err := meta.WriteTo(ctx, b); err != nil {
 		return nil, err
 	}
diff --git a/pkg/backend/filestate/meta_test.go b/pkg/backend/filestate/meta_test.go
index 1b97feb32b3f..a9437d255da1 100644
--- a/pkg/backend/filestate/meta_test.go
+++ b/pkg/backend/filestate/meta_test.go
@@ -26,9 +26,8 @@ import (
 	"gocloud.dev/blob/memblob"
 )
 
+//nolint:paralleltest // uses t.Setenv
 func TestEnsurePulumiMeta(t *testing.T) {
-	t.Parallel()
-
 	tests := []struct {
 		desc string
 		give map[string]string // files in the bucket
@@ -39,6 +38,37 @@ func TestEnsurePulumiMeta(t *testing.T) {
 			// Empty bucket should be initialized to
 			// the current version by default.
 			desc: "empty",
+			want: pulumiMeta{Version: 1},
+		},
+		{
+			// Use legacy mode even for the new bucket
+			// because the environment variable is "1".
+			desc: "empty/legacy",
+			env:  map[string]string{PulumiFilestateLegacyLayoutEnvVar: "1"},
+			want: pulumiMeta{Version: 0},
+		},
+		{
+			// Use legacy mode even for the new bucket
+			// because the environment variable is "true".
+			desc: "empty/legacy/true",
+			env:  map[string]string{PulumiFilestateLegacyLayoutEnvVar: "true"},
+			want: pulumiMeta{Version: 0},
+		},
+		{
+			// Legacy mode is disabled by setting the env var
+			// to "false".
+			// This is also the default behavior.
+			desc: "empty/legacy/false",
+			env:  map[string]string{PulumiFilestateLegacyLayoutEnvVar: "false"},
+			want: pulumiMeta{Version: 1},
+		},
+		{
+			// Non-empty bucket without a version file
+			// should get version 0 for legacy mode.
+			desc: "legacy",
+			give: map[string]string{
+				".pulumi/stacks/a.json": `{}`,
+			},
 			want: pulumiMeta{Version: 0},
 		},
 		{
@@ -48,6 +78,15 @@ func TestEnsurePulumiMeta(t *testing.T) {
 			},
 			want: pulumiMeta{Version: 0},
 		},
+		{
+			// Non-empty bucket with a version file
+			// should get whatever is in the file.
+			desc: "version 1",
+			give: map[string]string{
+				".pulumi/meta.yaml": `version: 1`,
+			},
+			want: pulumiMeta{Version: 1},
+		},
 		{
 			desc: "future version",
 			give: map[string]string{
@@ -60,7 +99,9 @@ func TestEnsurePulumiMeta(t *testing.T) {
 	for _, tt := range tests {
 		tt := tt
 		t.Run(tt.desc, func(t *testing.T) {
-			t.Parallel()
+			for k, v := range tt.env {
+				t.Setenv(k, v)
+			}
 
 			b := memblob.OpenBucket(nil)
 			ctx := context.Background()
@@ -124,6 +165,7 @@ func TestMeta_roundTrip(t *testing.T) {
 		give pulumiMeta
 	}{
 		{desc: "zero", give: pulumiMeta{Version: 0}},
+		{desc: "one", give: pulumiMeta{Version: 1}},
 		{desc: "future", give: pulumiMeta{Version: 42}},
 	}
 
@@ -133,6 +175,11 @@ func TestMeta_roundTrip(t *testing.T) {
 			t.Parallel()
 
 			b := memblob.OpenBucket(nil)
+			// The bucket is always non-empty,
+			// so we won't automatically try to use version 1.
+			require.NoError(t,
+				b.WriteAll(context.Background(), ".pulumi/foo", []byte("bar"), nil))
+
 			ctx := context.Background()
 			require.NoError(t, tt.give.WriteTo(ctx, b))
 
@@ -159,13 +206,19 @@ func TestMeta_WriteTo_zero(t *testing.T) {
 	assert.NoFileExists(t, filepath.Join(tmpDir, ".pulumi", "meta.yaml"))
 }
 
-// Verify that we don't create a metadata file with version 0 in new buckets.
+// Verify that we don't create a metadata file with version 0 in buckets
+// that have other files.
 func TestNew_noMetaOnInit(t *testing.T) {
 	t.Parallel()
 
 	tmpDir := t.TempDir()
+	bucket, err := fileblob.OpenBucket(tmpDir, nil)
+	require.NoError(t, err)
+	require.NoError(t,
+		bucket.WriteAll(context.Background(), ".pulumi/foo", []byte("bar"), nil))
+
 	ctx := context.Background()
-	_, err := New(ctx, diagtest.LogSink(t), "file://"+filepath.ToSlash(tmpDir), nil)
+	_, err = New(ctx, diagtest.LogSink(t), "file://"+filepath.ToSlash(tmpDir), nil)
 	require.NoError(t, err)
 
 	assert.NoFileExists(t, filepath.Join(tmpDir, ".pulumi", "meta.yaml"))
diff --git a/pkg/backend/filestate/state.go b/pkg/backend/filestate/state.go
index 985a2e31edc3..a14b8f7177ae 100644
--- a/pkg/backend/filestate/state.go
+++ b/pkg/backend/filestate/state.go
@@ -16,6 +16,7 @@ package filestate
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"io"
 	"os"
@@ -127,7 +128,7 @@ func (b *localBackend) getTarget(
 	}
 	return &deploy.Target{
 		Name:         stack.Name(),
-		Organization: "", // filestate has no organizations
+		Organization: "organization", // filestate has no organizations really, but we just always say it's "organization"
 		Config:       cfg,
 		Decrypter:    dec,
 		Snapshot:     snapshot,
@@ -553,3 +554,26 @@ func (b *localBackend) addToHistory(ref *localBackendReference, update backend.U
 	checkpointFile := fmt.Sprintf("%s.checkpoint.%s", pathPrefix, ext)
 	return b.bucket.Copy(context.TODO(), checkpointFile, b.stackPath(ref), nil)
 }
+
+// isPulumiDirEmpty reports whether the .pulumi directory inside the bucket
+// (used by us for bookkeeping) is empty.
+// This will ignore files in the bucket outside of the .pulumi directory.
+func isPulumiDirEmpty(ctx context.Context, b Bucket) (bool, error) {
+	iter := b.List(&blob.ListOptions{
+		Delimiter: "/",
+		Prefix:    workspace.BookkeepingDir,
+	})
+
+	if _, err := iter.Next(ctx); err != nil {
+		if errors.Is(err, io.EOF) {
+			return true, nil
+		}
+		// io.EOF is expected if the bucket is empty
+		// but all other errors are not.
+		return false, fmt.Errorf("list bucket: %w", err)
+	}
+
+	// If we get here, iter.Next succeeded,
+	// so the bucket is not empty.
+	return false, nil
+}
diff --git a/pkg/backend/filestate/state_test.go b/pkg/backend/filestate/state_test.go
new file mode 100644
index 000000000000..8917784e7aa2
--- /dev/null
+++ b/pkg/backend/filestate/state_test.go
@@ -0,0 +1,68 @@
+package filestate
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"gocloud.dev/blob/memblob"
+)
+
+func TestIsPulumiDirEmpty(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		desc string
+		// List of files that exist in the bucket.
+		files []string
+
+		// Whether the pulumi directory is considered empty.
+		empty bool
+	}{
+		{
+			desc:  "empty",
+			empty: true,
+		},
+		{
+			desc: "non-state files",
+			files: []string{
+				"foo",
+				"bar",
+			},
+			empty: true,
+		},
+		{
+			desc: "state files",
+			files: []string{
+				".pulumi/stacks/foo.json",
+				".pulumi/stacks/bar.json",
+			},
+			empty: false,
+		},
+		{
+			desc: "has pulumi meta file",
+			files: []string{
+				".pulumi/meta.yaml",
+			},
+			empty: false,
+		},
+	}
+
+	for _, tt := range tests {
+		tt := tt
+		t.Run(tt.desc, func(t *testing.T) {
+			t.Parallel()
+
+			b := memblob.OpenBucket(nil)
+			ctx := context.Background()
+			for _, f := range tt.files {
+				require.NoError(t, b.WriteAll(ctx, f, []byte{}, nil))
+			}
+
+			got, err := isPulumiDirEmpty(ctx, b)
+			require.NoError(t, err)
+			assert.Equal(t, tt.empty, got)
+		})
+	}
+}
diff --git a/pkg/backend/filestate/store.go b/pkg/backend/filestate/store.go
index cf736d09e8fe..faf96bf8b135 100644
--- a/pkg/backend/filestate/store.go
+++ b/pkg/backend/filestate/store.go
@@ -15,12 +15,14 @@
 package filestate
 
 import (
+	"errors"
 	"fmt"
 	"path/filepath"
 	"strings"
 
 	"github.com/pulumi/pulumi/sdk/v3/go/common/encoding"
 	"github.com/pulumi/pulumi/sdk/v3/go/common/tokens"
+	"github.com/pulumi/pulumi/sdk/v3/go/common/util/contract"
 	"github.com/pulumi/pulumi/sdk/v3/go/common/util/fsutil"
 	"github.com/pulumi/pulumi/sdk/v3/go/common/workspace"
 )
@@ -73,6 +75,207 @@ type referenceStore interface {
 
 	// ParseReference parses a localBackendReference from a string.
 	ParseReference(ref string) (*localBackendReference, error)
+
+	// ValidateReference verifies that the provided reference is valid
+	// returning an error if it is not.
+	ValidateReference(*localBackendReference) error
+}
+
+// projectReferenceStore is a referenceStore that stores stack
+// information with the new project-based layout.
+//
+// This is version 1 of the stack storage format.
+type projectReferenceStore struct {
+	bucket Bucket
+
+	// currentProject is a thread-safe way to get the current project.
+	currentProject func() *workspace.Project
+}
+
+var _ referenceStore = (*projectReferenceStore)(nil)
+
+func newProjectReferenceStore(bucket Bucket, currentProject func() *workspace.Project) *projectReferenceStore {
+	return &projectReferenceStore{
+		bucket:         bucket,
+		currentProject: currentProject,
+	}
+}
+
+// newReference builds a new localBackendReference with the provided arguments.
+// This DOES NOT modify the underlying storage.
+func (p *projectReferenceStore) newReference(project, name tokens.Name) *localBackendReference {
+	return &localBackendReference{
+		name:           name,
+		project:        project,
+		store:          p,
+		currentProject: p.currentProject,
+	}
+}
+
+func (p *projectReferenceStore) StackBasePath(ref *localBackendReference) string {
+	contract.Requiref(ref.project != "", "ref.project", "must not be empty")
+	return filepath.Join(StacksDir, fsutil.NamePath(ref.project), fsutil.NamePath(ref.name))
+}
+
+func (p *projectReferenceStore) HistoryDir(stack *localBackendReference) string {
+	contract.Requiref(stack.project != "", "ref.project", "must not be empty")
+	return filepath.Join(HistoriesDir, fsutil.NamePath(stack.project), fsutil.NamePath(stack.name))
+}
+
+func (p *projectReferenceStore) BackupDir(stack *localBackendReference) string {
+	contract.Requiref(stack.project != "", "ref.project", "must not be empty")
+	return filepath.Join(BackupsDir, fsutil.NamePath(stack.project), fsutil.NamePath(stack.name))
+}
+
+func (p *projectReferenceStore) ParseReference(stackRef string) (*localBackendReference, error) {
+	// We accept the following forms:
+	//
+	// 1. <stack-name>
+	// 2. <org-name>/<stack-name>
+	// 3. <org-name>/<project-name>/<stack-name>
+	//
+	// org-name must always be "organization".
+	// This matches the behavior of the Pulumi Service storage backend.
+	if stackRef == "" {
+		return nil, errors.New("stack name must not be empty")
+	}
+
+	var name, project, org string
+	split := strings.Split(stackRef, "/") // guaranteed to have at least one element
+	switch len(split) {
+	case 1:
+		name = split[0]
+	case 2:
+		org = split[0]
+		name = split[1]
+	case 3:
+		org = split[0]
+		project = split[1]
+		name = split[2]
+	}
+
+	// If the provided stack name didn't include the org or project,
+	// infer them from the local environment.
+	if org == "" {
+		// Filestate organization MUST always be "organization"
+		org = "organization"
+	}
+
+	if org != "organization" {
+		return nil, errors.New("organization name must be 'organization'")
+	}
+
+	if project == "" {
+		currentProject := p.currentProject()
+		if currentProject == nil {
+			return nil, fmt.Errorf("if you're using the --stack flag, " +
+				"pass the fully qualified name (organization/project/stack)")
+		}
+
+		project = currentProject.Name.String()
+	}
+
+	if len(project) > 100 {
+		return nil, errors.New("project names are limited to 100 characters")
+	}
+
+	if project != "" && !tokens.IsName(project) {
+		return nil, fmt.Errorf(
+			"project names may only contain alphanumerics, hyphens, underscores, and periods: %s",
+			project)
+	}
+
+	if !tokens.IsName(name) || len(name) > 100 {
+		return nil, fmt.Errorf(
+			"stack names are limited to 100 characters and may only contain alphanumeric, hyphens, underscores, or periods: %s",
+			name)
+	}
+
+	return p.newReference(tokens.Name(project), tokens.Name(name)), nil
+}
+
+func (p *projectReferenceStore) ValidateReference(ref *localBackendReference) error {
+	if ref.project == "" {
+		return fmt.Errorf("bad stack reference, project was not set")
+	}
+	return nil
+}
+
+func (p *projectReferenceStore) ListProjects() ([]tokens.Name, error) {
+	path := StacksDir
+
+	files, err := listBucket(p.bucket, path)
+	if err != nil {
+		return nil, fmt.Errorf("error listing stacks: %w", err)
+	}
+
+	projects := make([]tokens.Name, 0, len(files))
+	for _, file := range files {
+		if !file.IsDir {
+			continue // ignore files
+		}
+
+		projName := objectName(file)
+		if !tokens.IsName(projName) {
+			// If this isn't a valid Name
+			// it won't be a project directory,
+			// so skip it.
+			continue
+		}
+
+		projects = append(projects, tokens.Name(projName))
+	}
+
+	return projects, nil
+}
+
+func (p *projectReferenceStore) ListReferences() ([]*localBackendReference, error) {
+	// The first level of the bucket is the project name.
+	// The second level of the bucket is the stack name.
+	path := StacksDir
+
+	projects, err := p.ListProjects()
+	if err != nil {
+		return nil, err
+	}
+
+	var stacks []*localBackendReference
+	for _, projName := range projects {
+		// TODO: Could we improve the efficiency here by firstly making listBucket return an enumerator not
+		// eagerly collecting all keys into a slice, and secondly by getting listBucket to return all
+		// descendent items not just the immediate children. We could then do the necessary splitting by
+		// file paths here to work out project names.
+		projectFiles, err := listBucket(p.bucket, filepath.Join(path, projName.String()))
+		if err != nil {
+			return nil, fmt.Errorf("error listing stacks: %w", err)
+		}
+
+		for _, projectFile := range projectFiles {
+			// Can ignore directories at this level
+			if projectFile.IsDir {
+				continue
+			}
+
+			objName := objectName(projectFile)
+			// Skip files without valid extensions (e.g., *.bak files).
+			ext := filepath.Ext(objName)
+			// But accept gzip compression
+			if ext == encoding.GZIPExt {
+				objName = strings.TrimSuffix(objName, encoding.GZIPExt)
+				ext = filepath.Ext(objName)
+			}
+
+			if _, has := encoding.Marshalers[ext]; !has {
+				continue
+			}
+
+			// Read in this stack's information.
+			name := objName[:len(objName)-len(ext)]
+			stacks = append(stacks, p.newReference(projName, tokens.Name(name)))
+		}
+	}
+
+	return stacks, nil
 }
 
 // legacyReferenceStore is a referenceStore that stores stack
@@ -103,14 +306,17 @@ func (p *legacyReferenceStore) newReference(name tokens.Name) *localBackendRefer
 }
 
 func (p *legacyReferenceStore) StackBasePath(ref *localBackendReference) string {
+	contract.Requiref(ref.project == "", "ref.project", "must be empty")
 	return filepath.Join(StacksDir, fsutil.NamePath(ref.name))
 }
 
 func (p *legacyReferenceStore) HistoryDir(stack *localBackendReference) string {
+	contract.Requiref(stack.project == "", "ref.project", "must be empty")
 	return filepath.Join(HistoriesDir, fsutil.NamePath(stack.name))
 }
 
 func (p *legacyReferenceStore) BackupDir(stack *localBackendReference) string {
+	contract.Requiref(stack.project == "", "ref.project", "must be empty")
 	return filepath.Join(BackupsDir, fsutil.NamePath(stack.name))
 }
 
@@ -123,6 +329,13 @@ func (p *legacyReferenceStore) ParseReference(stackRef string) (*localBackendRef
 	return p.newReference(tokens.Name(stackRef)), nil
 }
 
+func (p *legacyReferenceStore) ValidateReference(ref *localBackendReference) error {
+	if ref.project != "" {
+		return fmt.Errorf("bad stack reference, project was set")
+	}
+	return nil
+}
+
 func (p *legacyReferenceStore) ListReferences() ([]*localBackendReference, error) {
 	files, err := listBucket(p.bucket, StacksDir)
 	if err != nil {
diff --git a/pkg/backend/filestate/store_test.go b/pkg/backend/filestate/store_test.go
index 33c1ff94639a..afc1e032b9a3 100644
--- a/pkg/backend/filestate/store_test.go
+++ b/pkg/backend/filestate/store_test.go
@@ -20,6 +20,7 @@ import (
 	"testing"
 
 	"github.com/pulumi/pulumi/sdk/v3/go/common/tokens"
+	"github.com/pulumi/pulumi/sdk/v3/go/common/workspace"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"gocloud.dev/blob/memblob"
@@ -41,6 +42,82 @@ func TestLegacyReferenceStore_referencePaths(t *testing.T) {
 	assert.Equal(t, ".pulumi/backups/foo", ref.BackupDir())
 }
 
+func TestProjectReferenceStore_referencePaths(t *testing.T) {
+	t.Parallel()
+
+	bucket := memblob.OpenBucket(nil)
+	store := newProjectReferenceStore(bucket, func() *workspace.Project {
+		return &workspace.Project{Name: "test"}
+	})
+
+	ref, err := store.ParseReference("organization/myproject/mystack")
+	require.NoError(t, err)
+
+	assert.Equal(t, ".pulumi/stacks/myproject/mystack", ref.StackBasePath())
+	assert.Equal(t, ".pulumi/history/myproject/mystack", ref.HistoryDir())
+	assert.Equal(t, ".pulumi/backups/myproject/mystack", ref.BackupDir())
+}
+
+func TestProjectReferenceStore_ParseReference(t *testing.T) {
+	t.Parallel()
+
+	bucket := memblob.OpenBucket(nil)
+	store := newProjectReferenceStore(bucket, func() *workspace.Project {
+		return &workspace.Project{Name: "currentProject"}
+	})
+
+	tests := []struct {
+		desc string
+		give string
+
+		fqname  tokens.QName
+		name    tokens.Name
+		project tokens.Name
+		str     string
+	}{
+		{
+			desc:    "simple",
+			give:    "foo",
+			fqname:  "organization/currentProject/foo",
+			name:    "foo",
+			project: "currentProject",
+			str:     "foo",
+			// truncated because project name is the same as current project
+		},
+		{
+			desc:    "organization",
+			give:    "organization/foo",
+			fqname:  "organization/currentProject/foo",
+			name:    "foo",
+			project: "currentProject",
+			str:     "foo",
+		},
+		{
+			desc:    "fully qualified",
+			give:    "organization/project/foo",
+			fqname:  "organization/project/foo",
+			name:    "foo",
+			project: "project",
+			str:     "organization/project/foo", // doesn't match current project
+		},
+	}
+
+	for _, tt := range tests {
+		tt := tt
+		t.Run(tt.desc, func(t *testing.T) {
+			t.Parallel()
+
+			ref, err := store.ParseReference(tt.give)
+			require.NoError(t, err)
+
+			assert.Equal(t, tt.fqname, ref.FullyQualifiedName())
+			assert.Equal(t, tt.name, ref.Name())
+			assert.Equal(t, tt.project, ref.Project())
+			assert.Equal(t, tt.str, ref.String())
+		})
+	}
+}
+
 func TestLegacyReferenceStore_ParseReference_errors(t *testing.T) {
 	t.Parallel()
 
@@ -73,6 +150,69 @@ func TestLegacyReferenceStore_ParseReference_errors(t *testing.T) {
 	}
 }
 
+func TestProjectReferenceStore_ParseReference_errors(t *testing.T) {
+	t.Parallel()
+
+	bucket := memblob.OpenBucket(nil)
+	store := newProjectReferenceStore(bucket, func() *workspace.Project {
+		return nil // current project is not set
+	})
+
+	tests := []struct {
+		desc    string
+		give    string
+		wantErr string
+	}{
+		{
+			desc:    "empty",
+			wantErr: "must not be empty",
+		},
+		{
+			desc:    "bad organization",
+			give:    "foo/bar/baz",
+			wantErr: "organization name must be 'organization'",
+		},
+		{
+			desc:    "long project name",
+			give:    "organization/" + strings.Repeat("a", 101) + "/foo",
+			wantErr: "project names are limited to 100 characters",
+		},
+		{
+			desc:    "long project stack name",
+			give:    "organization/foo/" + strings.Repeat("a", 101),
+			wantErr: "stack names are limited to 100 characters",
+		},
+		{
+			desc:    "no current project",
+			give:    "organization/foo",
+			wantErr: "pass the fully qualified name",
+		},
+		{
+			desc:    "invalid project name",
+			give:    "organization/foo:bar/baz",
+			wantErr: "may only contain alphanumeric",
+		},
+		{
+			desc:    "invalid stack name",
+			give:    "organization/foo/baz:qux",
+			wantErr: "may only contain alphanumeric",
+		},
+	}
+
+	for _, tt := range tests {
+		tt := tt
+		t.Run(tt.desc, func(t *testing.T) {
+			t.Parallel()
+
+			require.NotEmpty(t, tt.wantErr,
+				"bad test case: wantErr must be non-empty")
+
+			_, err := store.ParseReference(tt.give)
+			assert.ErrorContains(t, err, tt.wantErr)
+		})
+	}
+}
+
 func TestLegacyReferenceStore_ListReferences(t *testing.T) {
 	t.Parallel()
 
@@ -149,3 +289,109 @@ func TestLegacyReferenceStore_ListReferences(t *testing.T) {
 		})
 	}
 }
+
+func TestProjectReferenceStore_List(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		desc string
+
+		// List of file paths relative to the storage root
+		// that should exist before ListReferences is called.
+		files []string
+
+		// List of fully-qualified stack names that should be returned
+		// by ListReferences.
+		stacks []tokens.QName
+
+		// List of project names that should be returned by ListProjects.
+		projects []tokens.Name
+	}{
+		{
+			desc:     "empty",
+			stacks:   []tokens.QName{},
+			projects: []tokens.Name{},
+		},
+		{
+			desc: "json",
+			files: []string{
+				".pulumi/stacks/proj/foo.json",
+			},
+			stacks:   []tokens.QName{"organization/proj/foo"},
+			projects: []tokens.Name{"proj"},
+		},
+		{
+			desc: "gzipped",
+			files: []string{
+				".pulumi/stacks/foo/bar.json.gz",
+			},
+			stacks:   []tokens.QName{"organization/foo/bar"},
+			projects: []tokens.Name{"foo"},
+		},
+		{
+			desc: "multiple",
+			files: []string{
+				".pulumi/stacks/a/foo.json",
+				".pulumi/stacks/b/bar.json.gz",
+				".pulumi/stacks/c/baz.json",
+			},
+			stacks: []tokens.QName{
+				"organization/a/foo",
+				"organization/b/bar",
+				"organization/c/baz",
+			},
+			projects: []tokens.Name{"a", "b", "c"},
+		},
+		{
+			desc: "extraneous files and directories",
+			files: []string{
+				".pulumi/stacks/a/foo.json",
+				".pulumi/stacks/foo.json",
+				".pulumi/stacks/bar/baz/qux.json", // nested too deep
+				".pulumi/stacks/a b/c.json",       // bad project name
+			},
+			stacks:   []tokens.QName{"organization/a/foo"},
+			projects: []tokens.Name{"a", "bar"},
+		},
+	}
+
+	for _, tt := range tests {
+		tt := tt
+		t.Run(tt.desc, func(t *testing.T) {
+			t.Parallel()
+
+			bucket := memblob.OpenBucket(nil)
+			store := newProjectReferenceStore(bucket, func() *workspace.Project {
+				return &workspace.Project{Name: "test"}
+			})
+
+			ctx := context.Background()
+			for _, f := range tt.files {
+				require.NoError(t, bucket.WriteAll(ctx, f, []byte{}, nil))
+			}
+
+			t.Run("Projects", func(t *testing.T) {
+				t.Parallel()
+
+				projects, err := store.ListProjects()
+				require.NoError(t, err)
+
+				assert.Equal(t, tt.projects, projects)
+			})
+
+			t.Run("References", func(t *testing.T) {
+				t.Parallel()
+
+				refs, err := store.ListReferences()
+				require.NoError(t, err)
+
+				got := make([]tokens.QName, len(refs))
+				for i, ref := range refs {
+					got[i] = ref.FullyQualifiedName()
+				}
+
+				assert.Equal(t, tt.stacks, got)
+			})
+		})
+	}
+}
diff --git a/pkg/cmd/pulumi/new_acceptance_test.go b/pkg/cmd/pulumi/new_acceptance_test.go
index 9897f6f10cfe..64fcdab6640d 100644
--- a/pkg/cmd/pulumi/new_acceptance_test.go
+++ b/pkg/cmd/pulumi/new_acceptance_test.go
@@ -149,7 +149,8 @@ func TestCreatingProjectWithPulumiBackendURL(t *testing.T) {
 	proj := loadProject(t, tempdir)
 	assert.Equal(t, defaultProjectName, proj.Name.String())
 	// Expect the stack directory to have a checkpoint file for the stack.
-	_, err = os.Stat(filepath.Join(fileStateDir, workspace.BookkeepingDir, workspace.StackDir, stackName+".json"))
+	_, err = os.Stat(filepath.Join(
+		fileStateDir, workspace.BookkeepingDir, workspace.StackDir, defaultProjectName, stackName+".json"))
 	assert.NoError(t, err)
 
 	b, err = currentBackend(ctx, nil, display.Options{})
diff --git a/sdk/go/common/env/env.go b/sdk/go/common/env/env.go
index e181a5845f33..0929a5af7b3a 100644
--- a/sdk/go/common/env/env.go
+++ b/sdk/go/common/env/env.go
@@ -68,3 +68,12 @@ fail without a --force parameter.`)
 
 var DebugGRPC = env.String("DEBUG_GRPC", `Enables debug tracing of Pulumi gRPC internals.
 The variable should be set to the log file to which gRPC debug traces will be sent.`)
+
+// Environment variables that affect the self-managed backend.
+var (
+	SelfManagedStateNoLegacyWarning = env.Bool("SELF_MANAGED_STATE_NO_LEGACY_WARNING",
+		"Disables the warning about legacy stack files mixed with project-scoped stack files.")
+
+	SelfManagedStateLegacyLayout = env.Bool("SELF_MANAGED_STATE_LEGACY_LAYOUT",
+		"Uses the legacy layout for new buckets, which currently default to project-scoped stacks.")
+)
diff --git a/tests/config_test.go b/tests/config_test.go
index ab9278a79da7..a5f9ba571488 100644
--- a/tests/config_test.go
+++ b/tests/config_test.go
@@ -71,10 +71,14 @@ func TestConfigCommands(t *testing.T) {
 
 		// check that the nested config does not exist because we didn't use path
 		_, stderr := e.RunCommandExpectError("pulumi", "config", "get", "outer")
-		assert.Equal(t, "error: configuration key 'outer' not found for stack 'test'", strings.Trim(stderr, "\r\n"))
+		assert.Equal(t,
+			"error: configuration key 'outer' not found for stack 'test'",
+			strings.Trim(stderr, "\r\n"))
 
 		_, stderr = e.RunCommandExpectError("pulumi", "config", "get", "myList")
-		assert.Equal(t, "error: configuration key 'myList' not found for stack 'test'", strings.Trim(stderr, "\r\n"))
+		assert.Equal(t,
+			"error: configuration key 'myList' not found for stack 'test'",
+			strings.Trim(stderr, "\r\n"))
 
 		// set the nested config using --path
 		e.RunCommand("pulumi", "config", "set-all", "--path",
diff --git a/tests/integration/integration_test.go b/tests/integration/integration_test.go
index a6b48d9978ae..5074ffc3f871 100644
--- a/tests/integration/integration_test.go
+++ b/tests/integration/integration_test.go
@@ -531,7 +531,7 @@ func TestDestroyStackRef(t *testing.T) {
 
 	e.RunCommand("pulumi", "up", "--skip-preview", "--yes")
 	e.CWD = os.TempDir()
-	e.RunCommand("pulumi", "destroy", "--skip-preview", "--yes", "-s", "dev")
+	e.RunCommand("pulumi", "destroy", "--skip-preview", "--yes", "-s", "organization/large_resource_js/dev")
 }
 
 //nolint:paralleltest // uses parallel programtest
diff --git a/tests/stack_test.go b/tests/stack_test.go
index 77dd04dda2b6..f29e0bc43436 100644
--- a/tests/stack_test.go
+++ b/tests/stack_test.go
@@ -367,7 +367,7 @@ func TestStackBackups(t *testing.T) {
 		const stackName = "imulup"
 
 		// Get the path to the backup directory for this project.
-		backupDir, err := getStackProjectBackupDir(e, stackName)
+		backupDir, err := getStackProjectBackupDir(e, "stack_outputs", stackName)
 		assert.NoError(t, err, "getting stack project backup path")
 		defer func() {
 			if !t.Failed() {
@@ -560,8 +560,8 @@ func TestLocalStateLocking(t *testing.T) {
 
 // stackFileFormatAsserters returns a function to assert that the current file
 // format is for gzip and plain formats respectively.
-func stackFileFormatAsserters(t *testing.T, e *ptesting.Environment, stackName string) (func(), func()) {
-	stacksDir := filepath.Join(".pulumi", "stacks")
+func stackFileFormatAsserters(t *testing.T, e *ptesting.Environment, projectName, stackName string) (func(), func()) {
+	stacksDir := filepath.Join(".pulumi", "stacks", projectName)
 	pathStack := filepath.Join(stacksDir, stackName+".json")
 	pathStackGzip := pathStack + ".gz"
 	pathStackBak := pathStack + ".bak"
@@ -622,7 +622,7 @@ func TestLocalStateGzip(t *testing.T) { //nolint:paralleltest
 	e.RunCommand("yarn", "install")
 	e.RunCommand("pulumi", "up", "--non-interactive", "--yes", "--skip-preview")
 
-	assertGzipFileFormat, assertPlainFileFormat := stackFileFormatAsserters(t, e, stackName)
+	assertGzipFileFormat, assertPlainFileFormat := stackFileFormatAsserters(t, e, "stack_dependencies", stackName)
 	switchGzipOff := func() { e.Setenv(filestate.PulumiFilestateGzipEnvVar, "0") }
 	switchGzipOn := func() { e.Setenv(filestate.PulumiFilestateGzipEnvVar, "1") }
 	pulumiUp := func() { e.RunCommand("pulumi", "up", "--non-interactive", "--yes", "--skip-preview") }
@@ -691,10 +691,11 @@ func assertBackupStackFile(t *testing.T, stackName string, file os.DirEntry, bef
 	assert.True(t, parsedTime < after, "False: %v < %v", parsedTime, after)
 }
 
-func getStackProjectBackupDir(e *ptesting.Environment, stackName string) (string, error) {
+func getStackProjectBackupDir(e *ptesting.Environment, projectName, stackName string) (string, error) {
 	return filepath.Join(e.RootPath,
 		workspace.BookkeepingDir,
 		workspace.BackupDir,
+		projectName,
 		stackName,
 	), nil
 }