diff --git a/agent/consul/tenancy_bridge_oss.go b/agent/consul/tenancy_bridge_oss.go
index 09004c2ba6..2272af74ce 100644
--- a/agent/consul/tenancy_bridge_oss.go
+++ b/agent/consul/tenancy_bridge_oss.go
@@ -6,6 +6,17 @@
 
 package consul
 
+func (b *V1TenancyBridge) PartitionExists(partition string) (bool, error) {
+	if partition == "default" {
+		return true, nil
+	}
+	return false, nil
+}
+
+func (b *V1TenancyBridge) IsPartitionMarkedForDeletion(partition string) (bool, error) {
+	return false, nil
+}
+
 func (b *V1TenancyBridge) NamespaceExists(partition, namespace string) (bool, error) {
 	if partition == "default" && namespace == "default" {
 		return true, nil
@@ -13,9 +24,6 @@ func (b *V1TenancyBridge) NamespaceExists(partition, namespace string) (bool, er
 	return false, nil
 }
 
-func (b *V1TenancyBridge) PartitionExists(partition string) (bool, error) {
-	if partition == "default" {
-		return true, nil
-	}
+func (b *V1TenancyBridge) IsNamespaceMarkedForDeletion(partition, namespace string) (bool, error) {
 	return false, nil
 }
diff --git a/agent/grpc-external/services/resource/delete.go b/agent/grpc-external/services/resource/delete.go
index 597d184d2d..459f97f28a 100644
--- a/agent/grpc-external/services/resource/delete.go
+++ b/agent/grpc-external/services/resource/delete.go
@@ -37,7 +37,7 @@ func (s *Server) Delete(ctx context.Context, req *pbresource.DeleteRequest) (*pb
 	}
 
 	// TODO(spatel): Refactor _ and entMeta in NET-4919
-	authz, _, err := s.getAuthorizer(tokenFromContext(ctx), acl.DefaultEnterpriseMeta())
+	authz, authzContext, err := s.getAuthorizer(tokenFromContext(ctx), acl.DefaultEnterpriseMeta())
 	if err != nil {
 		return nil, err
 	}
@@ -58,7 +58,7 @@ func (s *Server) Delete(ctx context.Context, req *pbresource.DeleteRequest) (*pb
 	}
 
 	// Check ACLs
-	err = reg.ACLs.Write(authz, existing)
+	err = reg.ACLs.Write(authz, authzContext, existing)
 	switch {
 	case acl.IsErrPermissionDenied(err):
 		return nil, status.Error(codes.PermissionDenied, err.Error())
diff --git a/agent/grpc-external/services/resource/mock_TenancyBridge.go b/agent/grpc-external/services/resource/mock_TenancyBridge.go
index f2dcc6e0b4..662b4004b9 100644
--- a/agent/grpc-external/services/resource/mock_TenancyBridge.go
+++ b/agent/grpc-external/services/resource/mock_TenancyBridge.go
@@ -9,6 +9,54 @@ type MockTenancyBridge struct {
 	mock.Mock
 }
 
+// IsNamespaceMarkedForDeletion provides a mock function with given fields: partition, namespace
+func (_m *MockTenancyBridge) IsNamespaceMarkedForDeletion(partition string, namespace string) (bool, error) {
+	ret := _m.Called(partition, namespace)
+
+	var r0 bool
+	var r1 error
+	if rf, ok := ret.Get(0).(func(string, string) (bool, error)); ok {
+		return rf(partition, namespace)
+	}
+	if rf, ok := ret.Get(0).(func(string, string) bool); ok {
+		r0 = rf(partition, namespace)
+	} else {
+		r0 = ret.Get(0).(bool)
+	}
+
+	if rf, ok := ret.Get(1).(func(string, string) error); ok {
+		r1 = rf(partition, namespace)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// IsPartitionMarkedForDeletion provides a mock function with given fields: partition
+func (_m *MockTenancyBridge) IsPartitionMarkedForDeletion(partition string) (bool, error) {
+	ret := _m.Called(partition)
+
+	var r0 bool
+	var r1 error
+	if rf, ok := ret.Get(0).(func(string) (bool, error)); ok {
+		return rf(partition)
+	}
+	if rf, ok := ret.Get(0).(func(string) bool); ok {
+		r0 = rf(partition)
+	} else {
+		r0 = ret.Get(0).(bool)
+	}
+
+	if rf, ok := ret.Get(1).(func(string) error); ok {
+		r1 = rf(partition)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
 // NamespaceExists provides a mock function with given fields: partition, namespace
 func (_m *MockTenancyBridge) NamespaceExists(partition string, namespace string) (bool, error) {
 	ret := _m.Called(partition, namespace)
diff --git a/agent/grpc-external/services/resource/read.go b/agent/grpc-external/services/resource/read.go
index 490909114b..ca248cd962 100644
--- a/agent/grpc-external/services/resource/read.go
+++ b/agent/grpc-external/services/resource/read.go
@@ -54,7 +54,7 @@ func (s *Server) Read(ctx context.Context, req *pbresource.ReadRequest) (*pbreso
 	}
 
 	// Check V1 tenancy exists for the V2 resource.
-	if err = v1TenancyExists(reg, s.V1TenancyBridge, req.Id.Tenancy); err != nil {
+	if err = v1TenancyExists(reg, s.V1TenancyBridge, req.Id.Tenancy, codes.NotFound); err != nil {
 		return nil, err
 	}
 
diff --git a/agent/grpc-external/services/resource/read_test.go b/agent/grpc-external/services/resource/read_test.go
index 82085b451f..4b93f210e3 100644
--- a/agent/grpc-external/services/resource/read_test.go
+++ b/agent/grpc-external/services/resource/read_test.go
@@ -161,22 +161,22 @@ func TestRead_Success(t *testing.T) {
 				"namespaced resource provides nonempty partition and namespace": func(artistId, recordLabelId *pbresource.ID) *pbresource.ID {
 					return artistId
 				},
-				"namespaced resource provides uppercase namespace and partition": func(artistId, _ *pbresource.ID) *pbresource.ID {
+				"namespaced resource provides uppercase partition and namespace": func(artistId, _ *pbresource.ID) *pbresource.ID {
 					id := clone(artistId)
 					id.Tenancy.Partition = strings.ToUpper(artistId.Tenancy.Partition)
 					id.Tenancy.Namespace = strings.ToUpper(artistId.Tenancy.Namespace)
 					return id
 				},
-				"namespaced resource inherits tokens namespace when empty": func(artistId, _ *pbresource.ID) *pbresource.ID {
-					id := clone(artistId)
-					id.Tenancy.Namespace = ""
-					return id
-				},
 				"namespaced resource inherits tokens partition when empty": func(artistId, _ *pbresource.ID) *pbresource.ID {
 					id := clone(artistId)
 					id.Tenancy.Partition = ""
 					return id
 				},
+				"namespaced resource inherits tokens namespace when empty": func(artistId, _ *pbresource.ID) *pbresource.ID {
+					id := clone(artistId)
+					id.Tenancy.Namespace = ""
+					return id
+				},
 				"namespaced resource inherits tokens partition and namespace when empty": func(artistId, _ *pbresource.ID) *pbresource.ID {
 					id := clone(artistId)
 					id.Tenancy.Partition = ""
diff --git a/agent/grpc-external/services/resource/server.go b/agent/grpc-external/services/resource/server.go
index e011c81c0f..64ac47e6dd 100644
--- a/agent/grpc-external/services/resource/server.go
+++ b/agent/grpc-external/services/resource/server.go
@@ -54,7 +54,9 @@ type ACLResolver interface {
 //go:generate mockery --name TenancyBridge --inpackage
 type TenancyBridge interface {
 	PartitionExists(partition string) (bool, error)
-	NamespaceExists(partition string, namespace string) (bool, error)
+	IsPartitionMarkedForDeletion(partition string) (bool, error)
+	NamespaceExists(partition, namespace string) (bool, error)
+	IsNamespaceMarkedForDeletion(partition, namespace string) (bool, error)
 }
 
 func NewServer(cfg Config) *Server {
@@ -145,14 +147,15 @@ func validateId(id *pbresource.ID, errorPrefix string) error {
 	return nil
 }
 
-func v1TenancyExists(reg *resource.Registration, v1Bridge TenancyBridge, tenancy *pbresource.Tenancy) error {
+// v1TenancyExists return an error with the passed in gRPC status code when tenancy partition or namespace do not exist.
+func v1TenancyExists(reg *resource.Registration, v1Bridge TenancyBridge, tenancy *pbresource.Tenancy, errCode codes.Code) error {
 	if reg.Scope == resource.ScopePartition || reg.Scope == resource.ScopeNamespace {
 		exists, err := v1Bridge.PartitionExists(tenancy.Partition)
 		switch {
 		case err != nil:
 			return err
 		case !exists:
-			return status.Errorf(codes.NotFound, "partition resource not found: %v", tenancy.Partition)
+			return status.Errorf(errCode, "partition resource not found: %v", tenancy.Partition)
 		}
 	}
 
@@ -162,7 +165,31 @@ func v1TenancyExists(reg *resource.Registration, v1Bridge TenancyBridge, tenancy
 		case err != nil:
 			return err
 		case !exists:
-			return status.Errorf(codes.NotFound, "namespace resource not found: %v", tenancy.Namespace)
+			return status.Errorf(errCode, "namespace resource not found: %v", tenancy.Namespace)
+		}
+	}
+	return nil
+}
+
+// v1TenancyMarkedForDeletion returns a gRPC InvalidArgument when either partition or namespace is marked for deletion.
+func v1TenancyMarkedForDeletion(reg *resource.Registration, v1Bridge TenancyBridge, tenancy *pbresource.Tenancy) error {
+	if reg.Scope == resource.ScopePartition || reg.Scope == resource.ScopeNamespace {
+		marked, err := v1Bridge.IsPartitionMarkedForDeletion(tenancy.Partition)
+		switch {
+		case err != nil:
+			return err
+		case marked:
+			return status.Errorf(codes.InvalidArgument, "partition marked for deletion: %v", tenancy.Partition)
+		}
+	}
+
+	if reg.Scope == resource.ScopeNamespace {
+		marked, err := v1Bridge.IsNamespaceMarkedForDeletion(tenancy.Partition, tenancy.Namespace)
+		switch {
+		case err != nil:
+			return err
+		case marked:
+			return status.Errorf(codes.InvalidArgument, "namespace marked for deletion: %v", tenancy.Namespace)
 		}
 	}
 	return nil
diff --git a/agent/grpc-external/services/resource/server_test.go b/agent/grpc-external/services/resource/server_test.go
index 0e14f73292..0530f51a2a 100644
--- a/agent/grpc-external/services/resource/server_test.go
+++ b/agent/grpc-external/services/resource/server_test.go
@@ -80,6 +80,8 @@ func testServer(t *testing.T) *Server {
 	mockTenancyBridge.On("NamespaceExists", resource.DefaultPartitionName, resource.DefaultNamespaceName).Return(true, nil)
 	mockTenancyBridge.On("PartitionExists", mock.Anything).Return(false, nil)
 	mockTenancyBridge.On("NamespaceExists", mock.Anything, mock.Anything).Return(false, nil)
+	mockTenancyBridge.On("IsPartitionMarkedForDeletion", resource.DefaultPartitionName).Return(false, nil)
+	mockTenancyBridge.On("IsNamespaceMarkedForDeletion", resource.DefaultPartitionName, resource.DefaultNamespaceName).Return(false, nil)
 
 	return NewServer(Config{
 		Logger:          testutil.Logger(t),
diff --git a/agent/grpc-external/services/resource/testing/testing.go b/agent/grpc-external/services/resource/testing/testing.go
index d6eb6c0692..b35a9491f3 100644
--- a/agent/grpc-external/services/resource/testing/testing.go
+++ b/agent/grpc-external/services/resource/testing/testing.go
@@ -72,6 +72,8 @@ func RunResourceServiceWithACL(t *testing.T, aclResolver svc.ACLResolver, regist
 	mockTenancyBridge := &svc.MockTenancyBridge{}
 	mockTenancyBridge.On("PartitionExists", resource.DefaultPartitionName).Return(true, nil)
 	mockTenancyBridge.On("NamespaceExists", resource.DefaultPartitionName, resource.DefaultNamespaceName).Return(true, nil)
+	mockTenancyBridge.On("IsPartitionMarkedForDeletion", resource.DefaultPartitionName).Return(false, nil)
+	mockTenancyBridge.On("IsNamespaceMarkedForDeletion", resource.DefaultPartitionName, resource.DefaultNamespaceName).Return(false, nil)
 
 	svc.NewServer(svc.Config{
 		Backend:         backend,
diff --git a/agent/grpc-external/services/resource/write.go b/agent/grpc-external/services/resource/write.go
index f55645ae94..b85b33ef23 100644
--- a/agent/grpc-external/services/resource/write.go
+++ b/agent/grpc-external/services/resource/write.go
@@ -37,23 +37,20 @@ import (
 var errUseWriteStatus = status.Error(codes.InvalidArgument, "resource.status can only be set using the WriteStatus endpoint")
 
 func (s *Server) Write(ctx context.Context, req *pbresource.WriteRequest) (*pbresource.WriteResponse, error) {
-	if err := validateWriteRequest(req); err != nil {
-		return nil, err
-	}
-
-	reg, err := s.resolveType(req.Resource.Id.Type)
+	reg, err := s.validateWriteRequest(req)
 	if err != nil {
 		return nil, err
 	}
 
-	// TODO(spatel): Refactor _ and entMeta as part of NET-4911
-	authz, _, err := s.getAuthorizer(tokenFromContext(ctx), acl.DefaultEnterpriseMeta())
+	v1EntMeta := v2TenancyToV1EntMeta(req.Resource.Id.Tenancy)
+	authz, authzContext, err := s.getAuthorizer(tokenFromContext(ctx), v1EntMeta)
 	if err != nil {
 		return nil, err
 	}
+	v1EntMetaToV2Tenancy(reg, v1EntMeta, req.Resource.Id.Tenancy)
 
-	// check acls
-	err = reg.ACLs.Write(authz, req.Resource)
+	// ACL check comes before tenancy existence checks to not leak tenancy "existence".
+	err = reg.ACLs.Write(authz, authzContext, req.Resource)
 	switch {
 	case acl.IsErrPermissionDenied(err):
 		return nil, status.Error(codes.PermissionDenied, err.Error())
@@ -73,6 +70,16 @@ func (s *Server) Write(ctx context.Context, req *pbresource.WriteRequest) (*pbre
 		)
 	}
 
+	// Check V1 tenancy exists for the V2 resource
+	if err = v1TenancyExists(reg, s.V1TenancyBridge, req.Resource.Id.Tenancy, codes.InvalidArgument); err != nil {
+		return nil, err
+	}
+
+	// Check V1 tenancy not marked for deletion.
+	if err = v1TenancyMarkedForDeletion(reg, s.V1TenancyBridge, req.Resource.Id.Tenancy); err != nil {
+		return nil, err
+	}
+
 	if err = reg.Mutate(req.Resource); err != nil {
 		return nil, status.Errorf(codes.Internal, "failed mutate hook: %v", err.Error())
 	}
@@ -258,7 +265,7 @@ func (s *Server) retryCAS(ctx context.Context, vsn string, cas func() error) err
 	return err
 }
 
-func validateWriteRequest(req *pbresource.WriteRequest) error {
+func (s *Server) validateWriteRequest(req *pbresource.WriteRequest) (*resource.Registration, error) {
 	var field string
 	switch {
 	case req.Resource == nil:
@@ -270,17 +277,34 @@ func validateWriteRequest(req *pbresource.WriteRequest) error {
 	}
 
 	if field != "" {
-		return status.Errorf(codes.InvalidArgument, "%s is required", field)
+		return nil, status.Errorf(codes.InvalidArgument, "%s is required", field)
 	}
 
 	if err := validateId(req.Resource.Id, "resource.id"); err != nil {
-		return err
+		return nil, err
 	}
 
 	if req.Resource.Owner != nil {
 		if err := validateId(req.Resource.Owner, "resource.owner"); err != nil {
-			return err
+			return nil, err
 		}
 	}
-	return nil
+
+	// Check type exists.
+	reg, err := s.resolveType(req.Resource.Id.Type)
+	if err != nil {
+		return nil, err
+	}
+
+	// Check scope
+	if reg.Scope == resource.ScopePartition && req.Resource.Id.Tenancy.Namespace != "" {
+		return nil, status.Errorf(
+			codes.InvalidArgument,
+			"partition scoped resource %s cannot have a namespace. got: %s",
+			resource.ToGVK(req.Resource.Id.Type),
+			req.Resource.Id.Tenancy.Namespace,
+		)
+	}
+
+	return reg, nil
 }
diff --git a/agent/grpc-external/services/resource/write_test.go b/agent/grpc-external/services/resource/write_test.go
index 8c886d9bf8..fbd1ed2b50 100644
--- a/agent/grpc-external/services/resource/write_test.go
+++ b/agent/grpc-external/services/resource/write_test.go
@@ -5,6 +5,7 @@ package resource
 
 import (
 	"context"
+	"strings"
 	"sync/atomic"
 	"testing"
 
@@ -16,11 +17,13 @@ import (
 	"google.golang.org/protobuf/types/known/anypb"
 
 	"github.com/hashicorp/consul/acl/resolver"
+	"github.com/hashicorp/consul/internal/resource"
 	"github.com/hashicorp/consul/internal/resource/demo"
 	"github.com/hashicorp/consul/internal/storage"
 	"github.com/hashicorp/consul/proto-public/pbresource"
 	pbdemov1 "github.com/hashicorp/consul/proto/private/pbdemo/v1"
 	pbdemov2 "github.com/hashicorp/consul/proto/private/pbdemo/v2"
+	"github.com/hashicorp/consul/proto/private/prototest"
 )
 
 func TestWrite_InputValidation(t *testing.T) {
@@ -29,49 +32,56 @@ func TestWrite_InputValidation(t *testing.T) {
 
 	demo.RegisterTypes(server.Registry)
 
-	testCases := map[string]func(*pbresource.WriteRequest){
-		"no resource": func(req *pbresource.WriteRequest) { req.Resource = nil },
-		"no id":       func(req *pbresource.WriteRequest) { req.Resource.Id = nil },
-		"no type":     func(req *pbresource.WriteRequest) { req.Resource.Id.Type = nil },
-		"no tenancy":  func(req *pbresource.WriteRequest) { req.Resource.Id.Tenancy = nil },
-		"no name":     func(req *pbresource.WriteRequest) { req.Resource.Id.Name = "" },
-		"no data":     func(req *pbresource.WriteRequest) { req.Resource.Data = nil },
-
-		// TODO(spatel): Refactor tenancy as part of NET-4911
-		//
-		// // clone necessary to not pollute DefaultTenancy
-		// "tenancy partition not default": func(req *pbresource.WriteRequest) {
-		// 	req.Resource.Id.Tenancy = clone(req.Resource.Id.Tenancy)
-		// 	req.Resource.Id.Tenancy.Partition = ""
-		// },
-		// "tenancy namespace not default": func(req *pbresource.WriteRequest) {
-		// 	req.Resource.Id.Tenancy = clone(req.Resource.Id.Tenancy)
-		// 	req.Resource.Id.Tenancy.Namespace = ""
-		// },
-		// "tenancy peername not local": func(req *pbresource.WriteRequest) {
-		// 	req.Resource.Id.Tenancy = clone(req.Resource.Id.Tenancy)
-		// 	req.Resource.Id.Tenancy.PeerName = ""
-		// },
-		"wrong data type": func(req *pbresource.WriteRequest) {
+	testCases := map[string]func(artist, recordLabel *pbresource.Resource) *pbresource.Resource{
+		"no resource": func(artist, recordLabel *pbresource.Resource) *pbresource.Resource { return nil },
+		"no id": func(artist, _ *pbresource.Resource) *pbresource.Resource {
+			artist.Id = nil
+			return artist
+		},
+		"no type": func(artist, _ *pbresource.Resource) *pbresource.Resource {
+			artist.Id.Type = nil
+			return artist
+		},
+		"no tenancy": func(artist, _ *pbresource.Resource) *pbresource.Resource {
+			artist.Id.Tenancy = nil
+			return artist
+		},
+		"no name": func(artist, _ *pbresource.Resource) *pbresource.Resource {
+			artist.Id.Name = ""
+			return artist
+		},
+		"no data": func(artist, _ *pbresource.Resource) *pbresource.Resource {
+			artist.Data = nil
+			return artist
+		},
+		"wrong data type": func(artist, _ *pbresource.Resource) *pbresource.Resource {
 			var err error
-			req.Resource.Data, err = anypb.New(&pbdemov2.Album{})
+			artist.Data, err = anypb.New(&pbdemov2.Album{})
 			require.NoError(t, err)
+			return artist
 		},
-		"fail validation hook": func(req *pbresource.WriteRequest) {
-			artist := &pbdemov2.Artist{}
-			require.NoError(t, req.Resource.Data.UnmarshalTo(artist))
-			artist.Name = "" // name cannot be empty
-			require.NoError(t, req.Resource.Data.MarshalFrom(artist))
+		"fail validation hook": func(artist, _ *pbresource.Resource) *pbresource.Resource {
+			buffer := &pbdemov2.Artist{}
+			require.NoError(t, artist.Data.UnmarshalTo(buffer))
+			buffer.Name = "" // name cannot be empty
+			require.NoError(t, artist.Data.MarshalFrom(buffer))
+			return artist
 		},
+		"partition scope with non-empty namespace": func(_, recordLabel *pbresource.Resource) *pbresource.Resource {
+			recordLabel.Id.Tenancy.Namespace = "bogus"
+			return recordLabel
+		},
+		// TODO(spatel): add cluster scope tests when we have an actual cluster scoped resource (e.g. partition)
 	}
 	for desc, modFn := range testCases {
 		t.Run(desc, func(t *testing.T) {
-			res, err := demo.GenerateV2Artist()
+			artist, err := demo.GenerateV2Artist()
 			require.NoError(t, err)
 
-			req := &pbresource.WriteRequest{Resource: res}
-			modFn(req)
+			recordLabel, err := demo.GenerateV1RecordLabel("LoonyTunes")
+			require.NoError(t, err)
 
+			req := &pbresource.WriteRequest{Resource: modFn(artist, recordLabel)}
 			_, err = client.Write(testContext(t), req)
 			require.Error(t, err)
 			require.Equal(t, codes.InvalidArgument.String(), status.Code(err).String())
@@ -102,31 +112,6 @@ func TestWrite_OwnerValidation(t *testing.T) {
 			modReqFn:      func(req *pbresource.WriteRequest) { req.Resource.Owner.Name = "" },
 			errorContains: "resource.owner.name",
 		},
-
-		// TODO(spatel): Refactor tenancy as part of NET-4911
-		//
-		// // clone necessary to not pollute DefaultTenancy
-		// "owner tenancy partition not default": {
-		// 	modReqFn: func(req *pbresource.WriteRequest) {
-		// 		req.Resource.Owner.Tenancy = clone(req.Resource.Owner.Tenancy)
-		// 		req.Resource.Owner.Tenancy.Partition = ""
-		// 	},
-		// 	errorContains: "resource.owner.tenancy.partition",
-		// },
-		// "owner tenancy namespace not default": {
-		// 	modReqFn: func(req *pbresource.WriteRequest) {
-		// 		req.Resource.Owner.Tenancy = clone(req.Resource.Owner.Tenancy)
-		// 		req.Resource.Owner.Tenancy.Namespace = ""
-		// 	},
-		// 	errorContains: "resource.owner.tenancy.namespace",
-		// },
-		// "owner tenancy peername not local": {
-		// 	modReqFn: func(req *pbresource.WriteRequest) {
-		// 		req.Resource.Owner.Tenancy = clone(req.Resource.Owner.Tenancy)
-		// 		req.Resource.Owner.Tenancy.PeerName = ""
-		// 	},
-		// 	errorContains: "resource.owner.tenancy.peername",
-		// },
 	}
 	for desc, tc := range testCases {
 		t.Run(desc, func(t *testing.T) {
@@ -227,20 +212,196 @@ func TestWrite_Mutate(t *testing.T) {
 	require.Equal(t, pbdemov2.Genre_GENRE_DISCO, artistData.Genre)
 }
 
-func TestWrite_ResourceCreation_Success(t *testing.T) {
-	server := testServer(t)
-	client := testClient(t, server)
+func TestWrite_Create_Success(t *testing.T) {
+	testCases := map[string]struct {
+		modFn           func(artist, recordLabel *pbresource.Resource) *pbresource.Resource
+		expectedTenancy *pbresource.Tenancy
+	}{
+		"namespaced resource provides nonempty partition and namespace": {
+			modFn: func(artist, _ *pbresource.Resource) *pbresource.Resource {
+				return artist
+			},
+			expectedTenancy: resource.DefaultNamespacedTenancy(),
+		},
+		"namespaced resource provides uppercase partition and namespace": {
+			modFn: func(artist, _ *pbresource.Resource) *pbresource.Resource {
+				artist.Id.Tenancy.Partition = strings.ToUpper(artist.Id.Tenancy.Partition)
+				artist.Id.Tenancy.Namespace = strings.ToUpper(artist.Id.Tenancy.Namespace)
+				return artist
+			},
+			expectedTenancy: resource.DefaultNamespacedTenancy(),
+		},
+		"namespaced resource inherits tokens partition when empty": {
+			modFn: func(artist, _ *pbresource.Resource) *pbresource.Resource {
+				artist.Id.Tenancy.Partition = ""
+				return artist
+			},
+			expectedTenancy: resource.DefaultNamespacedTenancy(),
+		},
+		"namespaced resource inherits tokens namespace when empty": {
+			modFn: func(artist, _ *pbresource.Resource) *pbresource.Resource {
+				artist.Id.Tenancy.Namespace = ""
+				return artist
+			},
+			expectedTenancy: resource.DefaultNamespacedTenancy(),
+		},
+		"namespaced resource inherits tokens partition and namespace when empty": {
+			modFn: func(artist, _ *pbresource.Resource) *pbresource.Resource {
+				artist.Id.Tenancy.Partition = ""
+				artist.Id.Tenancy.Namespace = ""
+				return artist
+			},
+			expectedTenancy: resource.DefaultNamespacedTenancy(),
+		},
+		"partitioned resource provides nonempty partition": {
+			modFn: func(_, recordLabel *pbresource.Resource) *pbresource.Resource {
+				return recordLabel
+			},
+			expectedTenancy: resource.DefaultPartitionedTenancy(),
+		},
+		"partitioned resource provides uppercase partition": {
+			modFn: func(_, recordLabel *pbresource.Resource) *pbresource.Resource {
+				recordLabel.Id.Tenancy.Partition = strings.ToUpper(recordLabel.Id.Tenancy.Partition)
+				return recordLabel
+			},
+			expectedTenancy: resource.DefaultPartitionedTenancy(),
+		},
+		"partitioned resource inherits tokens partition when empty": {
+			modFn: func(_, recordLabel *pbresource.Resource) *pbresource.Resource {
+				recordLabel.Id.Tenancy.Partition = ""
+				return recordLabel
+			},
+			expectedTenancy: resource.DefaultPartitionedTenancy(),
+		},
+		// TODO(spatel): Add cluster scope tests when we have an actual cluster scoped resource (e.g. partition)
+	}
+	for desc, tc := range testCases {
+		t.Run(desc, func(t *testing.T) {
+			server := testServer(t)
+			client := testClient(t, server)
+			demo.RegisterTypes(server.Registry)
 
-	demo.RegisterTypes(server.Registry)
+			recordLabel, err := demo.GenerateV1RecordLabel("LoonyTunes")
+			require.NoError(t, err)
 
-	res, err := demo.GenerateV2Artist()
-	require.NoError(t, err)
+			artist, err := demo.GenerateV2Artist()
+			require.NoError(t, err)
 
-	rsp, err := client.Write(testContext(t), &pbresource.WriteRequest{Resource: res})
-	require.NoError(t, err)
-	require.NotEmpty(t, rsp.Resource.Version, "resource should have version")
-	require.NotEmpty(t, rsp.Resource.Id.Uid, "resource id should have uid")
-	require.NotEmpty(t, rsp.Resource.Generation, "resource should have generation")
+			rsp, err := client.Write(testContext(t), &pbresource.WriteRequest{Resource: tc.modFn(artist, recordLabel)})
+			require.NoError(t, err)
+			require.NotEmpty(t, rsp.Resource.Version, "resource should have version")
+			require.NotEmpty(t, rsp.Resource.Id.Uid, "resource id should have uid")
+			require.NotEmpty(t, rsp.Resource.Generation, "resource should have generation")
+			prototest.AssertDeepEqual(t, tc.expectedTenancy, rsp.Resource.Id.Tenancy)
+		})
+	}
+}
+
+func TestWrite_Create_Invalid_Tenancy(t *testing.T) {
+	testCases := map[string]struct {
+		modFn       func(artist, recordLabel *pbresource.Resource) *pbresource.Resource
+		errCode     codes.Code
+		errContains string
+	}{
+		"namespaced resource provides nonexistant partition": {
+			modFn: func(artist, _ *pbresource.Resource) *pbresource.Resource {
+				artist.Id.Tenancy.Partition = "boguspartition"
+				return artist
+			},
+			errCode:     codes.InvalidArgument,
+			errContains: "partition",
+		},
+		"namespaced resource provides nonexistant namespace": {
+			modFn: func(artist, _ *pbresource.Resource) *pbresource.Resource {
+				artist.Id.Tenancy.Namespace = "bogusnamespace"
+				return artist
+			},
+			errCode:     codes.InvalidArgument,
+			errContains: "namespace",
+		},
+		"partitioned resource provides nonexistant partition": {
+			modFn: func(_, recordLabel *pbresource.Resource) *pbresource.Resource {
+				recordLabel.Id.Tenancy.Partition = "boguspartition"
+				return recordLabel
+			},
+			errCode:     codes.InvalidArgument,
+			errContains: "partition",
+		},
+	}
+	for desc, tc := range testCases {
+		t.Run(desc, func(t *testing.T) {
+			server := testServer(t)
+			client := testClient(t, server)
+			demo.RegisterTypes(server.Registry)
+
+			recordLabel, err := demo.GenerateV1RecordLabel("LoonyTunes")
+			require.NoError(t, err)
+
+			artist, err := demo.GenerateV2Artist()
+			require.NoError(t, err)
+
+			_, err = client.Write(testContext(t), &pbresource.WriteRequest{Resource: tc.modFn(artist, recordLabel)})
+			require.Error(t, err)
+			require.Equal(t, codes.InvalidArgument.String(), status.Code(err).String())
+			require.Contains(t, err.Error(), tc.errContains)
+		})
+	}
+}
+
+func TestWrite_Tenancy_MarkedForDeletion(t *testing.T) {
+	// Verify resource write fails when its partition or namespace is marked for deletion.
+	testCases := map[string]struct {
+		modFn       func(artist, recordLabel *pbresource.Resource, mockTenancyBridge *MockTenancyBridge) *pbresource.Resource
+		errContains string
+	}{
+		"namespaced resources partition marked for deletion": {
+			modFn: func(artist, _ *pbresource.Resource, mockTenancyBridge *MockTenancyBridge) *pbresource.Resource {
+				mockTenancyBridge.On("IsPartitionMarkedForDeletion", "part1").Return(true, nil)
+				return artist
+			},
+			errContains: "partition marked for deletion",
+		},
+		"namespaced resources namespace marked for deletion": {
+			modFn: func(artist, _ *pbresource.Resource, mockTenancyBridge *MockTenancyBridge) *pbresource.Resource {
+				mockTenancyBridge.On("IsPartitionMarkedForDeletion", "part1").Return(false, nil)
+				mockTenancyBridge.On("IsNamespaceMarkedForDeletion", "part1", "ns1").Return(true, nil)
+				return artist
+			},
+			errContains: "namespace marked for deletion",
+		},
+		"partitioned resources partition marked for deletion": {
+			modFn: func(_, recordLabel *pbresource.Resource, mockTenancyBridge *MockTenancyBridge) *pbresource.Resource {
+				mockTenancyBridge.On("IsPartitionMarkedForDeletion", "part1").Return(true, nil)
+				return recordLabel
+			},
+			errContains: "partition marked for deletion",
+		},
+	}
+	for desc, tc := range testCases {
+		t.Run(desc, func(t *testing.T) {
+			server := testServer(t)
+			client := testClient(t, server)
+			demo.RegisterTypes(server.Registry)
+			recordLabel, err := demo.GenerateV1RecordLabel("LoonyTunes")
+			require.NoError(t, err)
+			recordLabel.Id.Tenancy.Partition = "part1"
+
+			artist, err := demo.GenerateV2Artist()
+			require.NoError(t, err)
+			artist.Id.Tenancy.Partition = "part1"
+			artist.Id.Tenancy.Namespace = "ns1"
+
+			mockTenancyBridge := &MockTenancyBridge{}
+			mockTenancyBridge.On("PartitionExists", "part1").Return(true, nil)
+			mockTenancyBridge.On("NamespaceExists", "part1", "ns1").Return(true, nil)
+			server.V1TenancyBridge = mockTenancyBridge
+
+			_, err = client.Write(testContext(t), &pbresource.WriteRequest{Resource: tc.modFn(artist, recordLabel, mockTenancyBridge)})
+			require.Error(t, err)
+			require.Equal(t, codes.InvalidArgument.String(), status.Code(err).String())
+			require.Contains(t, err.Error(), tc.errContains)
+		})
+	}
 }
 
 func TestWrite_CASUpdate_Success(t *testing.T) {
diff --git a/internal/mesh/internal/types/proxy_state_template.go b/internal/mesh/internal/types/proxy_state_template.go
index 84195ed2c6..2637a53a6d 100644
--- a/internal/mesh/internal/types/proxy_state_template.go
+++ b/internal/mesh/internal/types/proxy_state_template.go
@@ -44,10 +44,10 @@ func RegisterProxyStateTemplate(r resource.Registry) {
 
 				return nil
 			},
-			Write: func(authorizer acl.Authorizer, p *pbresource.Resource) error {
+			Write: func(authorizer acl.Authorizer, authzContext *acl.AuthorizerContext, p *pbresource.Resource) error {
 				// Require operator:write only for "break-glass" scenarios as this resource should be mostly
 				// managed by a controller.
-				return authorizer.ToAllowAuthorizer().OperatorWriteAllowed(resource.AuthorizerContext(p.Id.Tenancy))
+				return authorizer.ToAllowAuthorizer().OperatorWriteAllowed(authzContext)
 			},
 			List: func(authorizer acl.Authorizer, tenancy *pbresource.Tenancy) error {
 				// No-op List permission as we want to default to filtering resources
diff --git a/internal/resource/demo/demo.go b/internal/resource/demo/demo.go
index 6fd79f0b6c..a7901dd128 100644
--- a/internal/resource/demo/demo.go
+++ b/internal/resource/demo/demo.go
@@ -85,7 +85,7 @@ func RegisterTypes(r resource.Registry) {
 		return authz.ToAllowAuthorizer().KeyReadAllowed(key, authzContext)
 	}
 
-	writeACL := func(authz acl.Authorizer, res *pbresource.Resource) error {
+	writeACL := func(authz acl.Authorizer, authzContext *acl.AuthorizerContext, res *pbresource.Resource) error {
 		key := fmt.Sprintf("resource/%s/%s", resource.ToGVK(res.Id.Type), res.Id.Name)
 		return authz.ToAllowAuthorizer().KeyWriteAllowed(key, &acl.AuthorizerContext{})
 	}
@@ -200,11 +200,9 @@ func GenerateV1RecordLabel(name string) (*pbresource.Resource, error) {
 
 	return &pbresource.Resource{
 		Id: &pbresource.ID{
-			Type: TypeV1RecordLabel,
-			Tenancy: &pbresource.Tenancy{
-				Partition: resource.DefaultPartitionName,
-			},
-			Name: name,
+			Type:    TypeV1RecordLabel,
+			Tenancy: resource.DefaultPartitionedTenancy(),
+			Name:    name,
 		},
 		Data: data,
 		Metadata: map[string]string{
@@ -236,7 +234,7 @@ func GenerateV2Artist() (*pbresource.Resource, error) {
 	return &pbresource.Resource{
 		Id: &pbresource.ID{
 			Type:    TypeV2Artist,
-			Tenancy: TenancyDefault,
+			Tenancy: resource.DefaultNamespacedTenancy(),
 			Name:    fmt.Sprintf("%s-%s", strings.ToLower(adjective), strings.ToLower(noun)),
 		},
 		Data: data,
@@ -279,7 +277,7 @@ func generateV2Album(artistID *pbresource.ID, rand *rand.Rand) (*pbresource.Reso
 	return &pbresource.Resource{
 		Id: &pbresource.ID{
 			Type:    TypeV2Album,
-			Tenancy: artistID.Tenancy,
+			Tenancy: clone(artistID.Tenancy),
 			Name:    fmt.Sprintf("%s/%s-%s", artistID.Name, strings.ToLower(adjective), strings.ToLower(noun)),
 		},
 		Owner: artistID,
diff --git a/internal/resource/registry.go b/internal/resource/registry.go
index afc3f25bcd..9ebb13f9c9 100644
--- a/internal/resource/registry.go
+++ b/internal/resource/registry.go
@@ -62,7 +62,7 @@ type ACLHooks struct {
 	// Write is used to authorize Write and Delete RPCs.
 	//
 	// If it is omitted, `operator:write` permission is assumed.
-	Write func(acl.Authorizer, *pbresource.Resource) error
+	Write func(acl.Authorizer, *acl.AuthorizerContext, *pbresource.Resource) error
 
 	// List is used to authorize List RPCs.
 	//
@@ -124,8 +124,8 @@ func (r *TypeRegistry) Register(registration Registration) {
 		}
 	}
 	if registration.ACLs.Write == nil {
-		registration.ACLs.Write = func(authz acl.Authorizer, id *pbresource.Resource) error {
-			return authz.ToAllowAuthorizer().OperatorWriteAllowed(&acl.AuthorizerContext{})
+		registration.ACLs.Write = func(authz acl.Authorizer, authzContext *acl.AuthorizerContext, id *pbresource.Resource) error {
+			return authz.ToAllowAuthorizer().OperatorWriteAllowed(authzContext)
 		}
 	}
 	if registration.ACLs.List == nil {
diff --git a/internal/resource/registry_test.go b/internal/resource/registry_test.go
index d717f46f59..9ab00f8414 100644
--- a/internal/resource/registry_test.go
+++ b/internal/resource/registry_test.go
@@ -47,8 +47,8 @@ func TestRegister_Defaults(t *testing.T) {
 	require.True(t, acl.IsErrPermissionDenied(reg.ACLs.Read(testutils.ACLNoPermissions(t), nil, artist.Id)))
 
 	// verify default write hook requires operator:write
-	require.NoError(t, reg.ACLs.Write(testutils.ACLOperatorWrite(t), artist))
-	require.True(t, acl.IsErrPermissionDenied(reg.ACLs.Write(testutils.ACLNoPermissions(t), artist)))
+	require.NoError(t, reg.ACLs.Write(testutils.ACLOperatorWrite(t), nil, artist))
+	require.True(t, acl.IsErrPermissionDenied(reg.ACLs.Write(testutils.ACLNoPermissions(t), nil, artist)))
 
 	// verify default list hook requires operator:read
 	require.NoError(t, reg.ACLs.List(testutils.ACLOperatorRead(t), artist.Id.Tenancy))
diff --git a/internal/resource/tenancy.go b/internal/resource/tenancy.go
index e783a004fc..fd29f61725 100644
--- a/internal/resource/tenancy.go
+++ b/internal/resource/tenancy.go
@@ -43,7 +43,7 @@ func (s Scope) String() string {
 	panic(fmt.Sprintf("string mapping missing for scope %v", int(s)))
 }
 
-// Normalize lowercases partition and namespace.
+// Normalize lowercases the partition and namespace.
 func Normalize(tenancy *pbresource.Tenancy) {
 	if tenancy == nil {
 		return
@@ -51,3 +51,30 @@ func Normalize(tenancy *pbresource.Tenancy) {
 	tenancy.Partition = strings.ToLower(tenancy.Partition)
 	tenancy.Namespace = strings.ToLower(tenancy.Namespace)
 }
+
+// DefaultClusteredTenancy returns the default tenancy for a cluster scoped resource.
+func DefaultClusteredTenancy() *pbresource.Tenancy {
+	return &pbresource.Tenancy{
+		// TODO(spatel): Remove as part of "peer is not part of tenancy" ADR
+		PeerName: "local",
+	}
+}
+
+// DefaultPartitionedTenancy returns the default tenancy for a partition scoped resource.
+func DefaultPartitionedTenancy() *pbresource.Tenancy {
+	return &pbresource.Tenancy{
+		Partition: DefaultPartitionName,
+		// TODO(spatel): Remove as part of "peer is not part of tenancy" ADR
+		PeerName: "local",
+	}
+}
+
+// DefaultNamespedTenancy returns the default tenancy for a namespace scoped resource.
+func DefaultNamespacedTenancy() *pbresource.Tenancy {
+	return &pbresource.Tenancy{
+		Partition: DefaultPartitionName,
+		Namespace: DefaultNamespaceName,
+		// TODO(spatel): Remove as part of "peer is not part of tenancy" ADR
+		PeerName: "local",
+	}
+}