diff --git a/gateway/resolver/relations.go b/gateway/resolver/relations.go
new file mode 100644
index 0000000..4c54854
--- /dev/null
+++ b/gateway/resolver/relations.go
@@ -0,0 +1,203 @@
+package resolver
+
+import (
+	"context"
+	"strings"
+
+	"github.com/graphql-go/graphql"
+	"golang.org/x/text/cases"
+	"golang.org/x/text/language"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+// referenceInfo holds extracted reference details
+type referenceInfo struct {
+	name      string
+	namespace string
+	kind      string
+	apiGroup  string
+}
+
+// RelationResolver creates a GraphQL resolver for relation fields
+// Relationships are only enabled for GetItem queries to prevent N+1 problems in ListItems and Subscriptions
+func (r *Service) RelationResolver(fieldName string, gvk schema.GroupVersionKind) graphql.FieldResolveFn {
+	return func(p graphql.ResolveParams) (interface{}, error) {
+		// Determine operation type from GraphQL path analysis
+		operation := r.detectOperationFromGraphQLInfo(p)
+
+		r.log.Debug().
+			Str("fieldName", fieldName).
+			Str("operation", operation).
+			Str("graphqlField", p.Info.FieldName).
+			Msg("RelationResolver called")
+
+		// Check if relationships are allowed in this query context
+		if !r.isRelationResolutionAllowedForOperation(operation) {
+			r.log.Debug().
+				Str("fieldName", fieldName).
+				Str("operation", operation).
+				Msg("Relationship resolution disabled for this operation type")
+			return nil, nil
+		}
+
+		parentObj, ok := p.Source.(map[string]any)
+		if !ok {
+			return nil, nil
+		}
+
+		refInfo := r.extractReferenceInfo(parentObj, fieldName)
+		if refInfo.name == "" {
+			return nil, nil
+		}
+
+		return r.resolveReference(p.Context, refInfo, gvk)
+	}
+}
+
+// extractReferenceInfo extracts reference details from a *Ref object
+func (r *Service) extractReferenceInfo(parentObj map[string]any, fieldName string) referenceInfo {
+	name, _ := parentObj["name"].(string)
+	if name == "" {
+		return referenceInfo{}
+	}
+
+	namespace, _ := parentObj["namespace"].(string)
+	apiGroup, _ := parentObj["apiGroup"].(string)
+
+	kind, _ := parentObj["kind"].(string)
+	if kind == "" {
+		// Fallback: infer kind from field name (e.g., "role" -> "Role")
+		kind = cases.Title(language.English).String(fieldName)
+	}
+
+	return referenceInfo{
+		name:      name,
+		namespace: namespace,
+		kind:      kind,
+		apiGroup:  apiGroup,
+	}
+}
+
+// resolveReference fetches a referenced Kubernetes resource using strict conflict resolution
+func (r *Service) resolveReference(ctx context.Context, ref referenceInfo, targetGVK schema.GroupVersionKind) (interface{}, error) {
+	// Use provided reference info to override GVK if specified
+	finalGVK := targetGVK
+	if ref.apiGroup != "" {
+		finalGVK.Group = ref.apiGroup
+	}
+	if ref.kind != "" {
+		finalGVK.Kind = ref.kind
+	}
+
+	// Convert sanitized group to original before calling the client
+	finalGVK.Group = r.getOriginalGroupName(finalGVK.Group)
+
+	obj := &unstructured.Unstructured{}
+	obj.SetGroupVersionKind(finalGVK)
+
+	key := client.ObjectKey{Name: ref.name}
+	if ref.namespace != "" {
+		key.Namespace = ref.namespace
+	}
+
+	if err := r.runtimeClient.Get(ctx, key, obj); err != nil {
+		// For "not found" errors, return nil to allow graceful degradation
+		// This handles cases where referenced resources are deleted or don't exist
+		if apierrors.IsNotFound(err) {
+			return nil, nil
+		}
+
+		// For other errors (network, permission, etc.), log and return the actual error
+		// This ensures proper error propagation for debugging and monitoring
+		r.log.Error().
+			Err(err).
+			Str("operation", "resolve_relation").
+			Str("group", finalGVK.Group).
+			Str("version", finalGVK.Version).
+			Str("kind", finalGVK.Kind).
+			Str("name", ref.name).
+			Str("namespace", ref.namespace).
+			Msg("Unable to resolve referenced object")
+		return nil, err
+	}
+
+	// Happy path: resource found successfully
+	return obj.Object, nil
+}
+
+// isRelationResolutionAllowedForOperation checks if relationship resolution should be enabled for the given operation type
+func (r *Service) isRelationResolutionAllowedForOperation(operation string) bool {
+	// Only allow relationships for GetItem and GetItemAsYAML operations
+	switch operation {
+	case GET_ITEM, GET_ITEM_AS_YAML:
+		return true
+	case LIST_ITEMS, SUBSCRIBE_ITEM, SUBSCRIBE_ITEMS:
+		return false
+	default:
+		// For unknown operations, be conservative and disable relationships
+		r.log.Debug().Str("operation", operation).Msg("Unknown operation type, disabling relationships")
+		return false
+	}
+}
+
+// detectOperationFromGraphQLInfo analyzes GraphQL field path to determine operation type
+// This looks at the parent field context to determine if we're in a list, single item, or subscription
+func (r *Service) detectOperationFromGraphQLInfo(p graphql.ResolveParams) string {
+	if p.Info.Path == nil {
+		return "unknown"
+	}
+
+	// Walk up the path to find the parent resolver context
+	path := p.Info.Path
+	for path.Prev != nil {
+		path = path.Prev
+
+		// Check if we find a parent field that indicates the operation type
+		if fieldName, ok := path.Key.(string); ok {
+			fieldLower := strings.ToLower(fieldName)
+
+			// Check for subscription patterns
+			if strings.Contains(fieldLower, "subscription") {
+				r.log.Debug().
+					Str("parentField", fieldName).
+					Msg("Detected subscription context from parent field")
+				return SUBSCRIBE_ITEMS
+			}
+
+			// Check for mutation patterns
+			if strings.HasPrefix(fieldLower, "create") {
+				return CREATE_ITEM
+			}
+			if strings.HasPrefix(fieldLower, "update") {
+				return UPDATE_ITEM
+			}
+			if strings.HasPrefix(fieldLower, "delete") {
+				return DELETE_ITEM
+			}
+
+			// Check for YAML patterns
+			if strings.HasSuffix(fieldLower, "yaml") {
+				return GET_ITEM_AS_YAML
+			}
+
+			// Check for list patterns (plural without args, or explicitly plural fields)
+			if strings.HasSuffix(fieldName, "s") && !strings.HasSuffix(fieldName, "Status") {
+				// This looks like a plural field, likely a list operation
+				r.log.Debug().
+					Str("parentField", fieldName).
+					Msg("Detected list context from parent field")
+				return LIST_ITEMS
+			}
+		}
+	}
+
+	// If we can't determine from parent context, assume it's a single item operation
+	// This is the safe default that allows relationships for queries
+	r.log.Debug().
+		Str("currentField", p.Info.FieldName).
+		Msg("Could not determine operation type from GraphQL path, defaulting to GetItem (enables relations)")
+	return GET_ITEM
+}
diff --git a/gateway/resolver/resolver.go b/gateway/resolver/resolver.go
index a99e38d..c8f813f 100644
--- a/gateway/resolver/resolver.go
+++ b/gateway/resolver/resolver.go
@@ -25,11 +25,23 @@ import (
 	"github.com/openmfp/golang-commons/logger"
 )
 
+const (
+	LIST_ITEMS       = "ListItems"
+	GET_ITEM         = "GetItem"
+	GET_ITEM_AS_YAML = "GetItemAsYAML"
+	CREATE_ITEM      = "CreateItem"
+	UPDATE_ITEM      = "UpdateItem"
+	DELETE_ITEM      = "DeleteItem"
+	SUBSCRIBE_ITEM   = "SubscribeItem"
+	SUBSCRIBE_ITEMS  = "SubscribeItems"
+)
+
 type Provider interface {
 	CrudProvider
 	CustomQueriesProvider
 	CommonResolver() graphql.FieldResolveFn
 	SanitizeGroupName(string) string
+	RelationResolver(fieldName string, gvk schema.GroupVersionKind) graphql.FieldResolveFn
 }
 
 type CrudProvider interface {
@@ -65,7 +77,7 @@ func New(log *logger.Logger, runtimeClient client.WithWatch) *Service {
 // ListItems returns a GraphQL CommonResolver function that lists Kubernetes resources of the given GroupVersionKind.
 func (r *Service) ListItems(gvk schema.GroupVersionKind, scope v1.ResourceScope) graphql.FieldResolveFn {
 	return func(p graphql.ResolveParams) (interface{}, error) {
-		ctx, span := otel.Tracer("").Start(p.Context, "ListItems", trace.WithAttributes(attribute.String("kind", gvk.Kind)))
+		ctx, span := otel.Tracer("").Start(p.Context, LIST_ITEMS, trace.WithAttributes(attribute.String("kind", gvk.Kind)))
 		defer span.End()
 
 		gvk.Group = r.getOriginalGroupName(gvk.Group)
@@ -87,6 +99,7 @@ func (r *Service) ListItems(gvk schema.GroupVersionKind, scope v1.ResourceScope)
 		list.SetGroupVersionKind(gvk)
 
 		var opts []client.ListOption
+
 		// Handle label selector argument
 		if labelSelector, ok := p.Args[LabelSelectorArg].(string); ok && labelSelector != "" {
 			selector, err := labels.Parse(labelSelector)
@@ -117,16 +130,16 @@ func (r *Service) ListItems(gvk schema.GroupVersionKind, scope v1.ResourceScope)
 			return nil, err
 		}
 
-		err = validateSortBy(list.Items, sortBy)
-		if err != nil {
-			log.Error().Err(err).Str(SortByArg, sortBy).Msg("Invalid sortBy field path")
-			return nil, err
+		if sortBy != "" {
+			if err := validateSortBy(list.Items, sortBy); err != nil {
+				log.Error().Err(err).Str(SortByArg, sortBy).Msg("Invalid sortBy field path")
+				return nil, err
+			}
+			sort.Slice(list.Items, func(i, j int) bool {
+				return compareUnstructured(list.Items[i], list.Items[j], sortBy) < 0
+			})
 		}
 
-		sort.Slice(list.Items, func(i, j int) bool {
-			return compareUnstructured(list.Items[i], list.Items[j], sortBy) < 0
-		})
-
 		items := make([]map[string]any, len(list.Items))
 		for i, item := range list.Items {
 			items[i] = item.Object
diff --git a/gateway/schema/relations.go b/gateway/schema/relations.go
new file mode 100644
index 0000000..e9dfa79
--- /dev/null
+++ b/gateway/schema/relations.go
@@ -0,0 +1,141 @@
+package schema
+
+import (
+	"strings"
+
+	"golang.org/x/text/cases"
+	"golang.org/x/text/language"
+
+	"github.com/go-openapi/spec"
+	"github.com/graphql-go/graphql"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+)
+
+// addRelationFields adds relation fields to schemas that contain *Ref fields
+func (g *Gateway) addRelationFields(fields graphql.Fields, properties map[string]spec.Schema) {
+	for fieldName := range properties {
+		if !strings.HasSuffix(fieldName, "Ref") {
+			continue
+		}
+
+		baseName := strings.TrimSuffix(fieldName, "Ref")
+		sanitizedFieldName := sanitizeFieldName(fieldName)
+
+		refField, exists := fields[sanitizedFieldName]
+		if !exists {
+			continue
+		}
+
+		enhancedType := g.enhanceRefTypeWithRelation(refField.Type, baseName)
+		if enhancedType == nil {
+			continue
+		}
+
+		fields[sanitizedFieldName] = &graphql.Field{
+			Type: enhancedType,
+		}
+	}
+}
+
+// enhanceRefTypeWithRelation adds a relation field to a *Ref object type
+func (g *Gateway) enhanceRefTypeWithRelation(originalType graphql.Output, baseName string) graphql.Output {
+	objType, ok := originalType.(*graphql.Object)
+	if !ok {
+		return originalType
+	}
+
+	cacheKey := objType.Name() + "_" + baseName + "_Enhanced"
+	if enhancedType, exists := g.enhancedTypesCache[cacheKey]; exists {
+		return enhancedType
+	}
+
+	enhancedFields := g.copyOriginalFields(objType.Fields())
+	g.addRelationField(enhancedFields, baseName)
+
+	enhancedType := graphql.NewObject(graphql.ObjectConfig{
+		Name:   sanitizeFieldName(cacheKey),
+		Fields: enhancedFields,
+	})
+
+	g.enhancedTypesCache[cacheKey] = enhancedType
+	return enhancedType
+}
+
+// copyOriginalFields converts FieldDefinition to Field for reuse
+func (g *Gateway) copyOriginalFields(originalFieldDefs graphql.FieldDefinitionMap) graphql.Fields {
+	enhancedFields := make(graphql.Fields, len(originalFieldDefs))
+	for fieldName, fieldDef := range originalFieldDefs {
+		enhancedFields[fieldName] = &graphql.Field{
+			Type:        fieldDef.Type,
+			Description: fieldDef.Description,
+			Resolve:     fieldDef.Resolve,
+		}
+	}
+	return enhancedFields
+}
+
+// addRelationField adds a single relation field to the enhanced fields
+func (g *Gateway) addRelationField(enhancedFields graphql.Fields, baseName string) {
+	targetType, targetGVK, ok := g.findRelationTarget(baseName)
+	if !ok {
+		return
+	}
+
+	sanitizedBaseName := sanitizeFieldName(baseName)
+	enhancedFields[sanitizedBaseName] = &graphql.Field{
+		Type:    targetType,
+		Resolve: g.resolver.RelationResolver(baseName, *targetGVK),
+	}
+}
+
+// findRelationTarget locates the GraphQL output type and its GVK for a relation target
+func (g *Gateway) findRelationTarget(baseName string) (graphql.Output, *schema.GroupVersionKind, bool) {
+	targetKind := cases.Title(language.English).String(baseName)
+
+	for defKey, defSchema := range g.definitions {
+		if g.matchesTargetKind(defSchema, targetKind) {
+			// Resolve or build the GraphQL type
+			var fieldType graphql.Output
+			if existingType, exists := g.typesCache[defKey]; exists {
+				fieldType = existingType
+			} else {
+				ft, _, err := g.convertSwaggerTypeToGraphQL(defSchema, defKey, []string{}, make(map[string]bool))
+				if err != nil {
+					continue
+				}
+				fieldType = ft
+			}
+
+			// Extract GVK from the schema definition
+			gvk, err := g.getGroupVersionKind(defKey)
+			if err != nil || gvk == nil {
+				continue
+			}
+
+			return fieldType, gvk, true
+		}
+	}
+
+	return nil, nil, false
+}
+
+// matchesTargetKind checks if a schema definition matches the target kind
+func (g *Gateway) matchesTargetKind(defSchema spec.Schema, targetKind string) bool {
+	gvkExt, ok := defSchema.Extensions["x-kubernetes-group-version-kind"]
+	if !ok {
+		return false
+	}
+
+	gvkSlice, ok := gvkExt.([]any)
+	if !ok || len(gvkSlice) == 0 {
+		return false
+	}
+
+	gvkMap, ok := gvkSlice[0].(map[string]any)
+	if !ok {
+		return false
+	}
+
+	kind, ok := gvkMap["kind"].(string)
+	return ok && kind == targetKind
+}
diff --git a/gateway/schema/schema.go b/gateway/schema/schema.go
index 9085ec2..8db274c 100644
--- a/gateway/schema/schema.go
+++ b/gateway/schema/schema.go
@@ -22,16 +22,13 @@ type Provider interface {
 }
 
 type Gateway struct {
-	log           *logger.Logger
-	resolver      resolver.Provider
-	graphqlSchema graphql.Schema
-
-	definitions spec.Definitions
-
-	// typesCache stores generated GraphQL object types(fields) to prevent redundant repeated generation.
-	typesCache map[string]*graphql.Object
-	// inputTypesCache stores generated GraphQL input object types(input fields) to prevent redundant repeated generation.
-	inputTypesCache map[string]*graphql.InputObject
+	log                *logger.Logger
+	resolver           resolver.Provider
+	graphqlSchema      graphql.Schema
+	definitions        spec.Definitions
+	typesCache         map[string]*graphql.Object
+	inputTypesCache    map[string]*graphql.InputObject
+	enhancedTypesCache map[string]*graphql.Object // Cache for enhanced *Ref types
 	// Prevents naming conflict in case of the same Kind name in different groups/versions
 	typeNameRegistry map[string]string // map[Kind]GroupVersion
 
@@ -41,13 +38,14 @@ type Gateway struct {
 
 func New(log *logger.Logger, definitions spec.Definitions, resolverProvider resolver.Provider) (*Gateway, error) {
 	g := &Gateway{
-		log:              log,
-		resolver:         resolverProvider,
-		definitions:      definitions,
-		typesCache:       make(map[string]*graphql.Object),
-		inputTypesCache:  make(map[string]*graphql.InputObject),
-		typeNameRegistry: make(map[string]string),
-		typeByCategory:   make(map[string][]resolver.TypeByCategory),
+		log:                log,
+		resolver:           resolverProvider,
+		definitions:        definitions,
+		typesCache:         make(map[string]*graphql.Object),
+		inputTypesCache:    make(map[string]*graphql.InputObject),
+		enhancedTypesCache: make(map[string]*graphql.Object),
+		typeNameRegistry:   make(map[string]string),
+		typeByCategory:     make(map[string][]resolver.TypeByCategory),
 	}
 
 	err := g.generateGraphqlSchema()
@@ -336,6 +334,9 @@ func (g *Gateway) generateGraphQLFields(resourceScheme *spec.Schema, typePrefix
 		}
 	}
 
+	// Add relation fields for any *Ref fields in this schema
+	g.addRelationFields(fields, resourceScheme.Properties)
+
 	return fields, inputFields, nil
 }
 
diff --git a/listener/pkg/apischema/builder.go b/listener/pkg/apischema/builder.go
index 81ef41d..dd8d155 100644
--- a/listener/pkg/apischema/builder.go
+++ b/listener/pkg/apischema/builder.go
@@ -30,18 +30,31 @@ var (
 	ErrCRDNoVersions        = errors.New("CRD has no versions defined")
 	ErrMarshalGVK           = errors.New("failed to marshal GVK extension")
 	ErrUnmarshalGVK         = errors.New("failed to unmarshal GVK extension")
+	ErrBuildKindRegistry    = errors.New("failed to build kind registry")
 )
 
 type SchemaBuilder struct {
-	schemas map[string]*spec.Schema
-	err     *multierror.Error
-	log     *logger.Logger
+	schemas           map[string]*spec.Schema
+	err               *multierror.Error
+	log               *logger.Logger
+	kindRegistry      map[GroupVersionKind]ResourceInfo // Changed: Use GVK as key for precise lookup
+	preferredVersions map[string]string                 // map[group/kind]preferredVersion
+}
+
+// ResourceInfo holds information about a resource for relationship resolution
+type ResourceInfo struct {
+	Group     string
+	Version   string
+	Kind      string
+	SchemaKey string
 }
 
 func NewSchemaBuilder(oc openapi.Client, preferredApiGroups []string, log *logger.Logger) *SchemaBuilder {
 	b := &SchemaBuilder{
-		schemas: make(map[string]*spec.Schema),
-		log:     log,
+		schemas:           make(map[string]*spec.Schema),
+		kindRegistry:      make(map[GroupVersionKind]ResourceInfo),
+		preferredVersions: make(map[string]string),
+		log:               log,
 	}
 
 	apiv3Paths, err := oc.Paths()
@@ -107,7 +120,7 @@ func (b *SchemaBuilder) WithScope(rm meta.RESTMapper) *SchemaBuilder {
 				Str("group", gvks[0].Group).
 				Str("version", gvks[0].Version).
 				Str("kind", gvks[0].Kind).
-				Msg("failed to determine if GVK is namespaced")
+				Msg("failed to get namespaced info for GVK")
 			continue
 		}
 
@@ -116,61 +129,409 @@ func (b *SchemaBuilder) WithScope(rm meta.RESTMapper) *SchemaBuilder {
 		} else {
 			schema.VendorExtensible.AddExtension(common.ScopeExtensionKey, apiextensionsv1.ClusterScoped)
 		}
-
 	}
-
 	return b
 }
 
 func (b *SchemaBuilder) WithCRDCategories(crd *apiextensionsv1.CustomResourceDefinition) *SchemaBuilder {
+	if crd == nil {
+		return b
+	}
+
 	categories := crd.Spec.Names.Categories
 	if len(categories) == 0 {
 		return b
 	}
+
 	gvk, err := getCRDGroupVersionKind(crd.Spec)
 	if err != nil {
 		b.err = multierror.Append(b.err, errors.Join(ErrGetCRDGVK, err))
 		return b
 	}
 
-	schema, ok := b.schemas[getOpenAPISchemaKey(*gvk)]
-	if !ok {
-		return b
-	}
-
-	schema.VendorExtensible.AddExtension(common.CategoriesExtensionKey, categories)
+	for _, v := range crd.Spec.Versions {
+		resourceKey := getOpenAPISchemaKey(metav1.GroupVersionKind{Group: gvk.Group, Version: v.Name, Kind: gvk.Kind})
+		resourceSchema, ok := b.schemas[resourceKey]
+		if !ok {
+			continue
+		}
 
+		resourceSchema.VendorExtensible.AddExtension(common.CategoriesExtensionKey, categories)
+		b.schemas[resourceKey] = resourceSchema
+	}
 	return b
 }
 
 func (b *SchemaBuilder) WithApiResourceCategories(list []*metav1.APIResourceList) *SchemaBuilder {
+	if len(list) == 0 {
+		return b
+	}
+
 	for _, apiResourceList := range list {
+		gv, err := runtimeSchema.ParseGroupVersion(apiResourceList.GroupVersion)
+		if err != nil {
+			b.err = multierror.Append(b.err, errors.Join(ErrParseGroupVersion, err))
+			continue
+		}
 		for _, apiResource := range apiResourceList.APIResources {
 			if apiResource.Categories == nil {
 				continue
 			}
+			gvk := metav1.GroupVersionKind{Group: gv.Group, Version: gv.Version, Kind: apiResource.Kind}
+			resourceKey := getOpenAPISchemaKey(gvk)
+			resourceSchema, ok := b.schemas[resourceKey]
+			if !ok {
+				continue
+			}
+			resourceSchema.VendorExtensible.AddExtension(common.CategoriesExtensionKey, apiResource.Categories)
+			b.schemas[resourceKey] = resourceSchema
+		}
+	}
+	return b
+}
+
+// WithPreferredVersions populates preferred version information from API discovery
+func (b *SchemaBuilder) WithPreferredVersions(apiResLists []*metav1.APIResourceList) *SchemaBuilder {
+	for _, apiResList := range apiResLists {
+		gv, err := runtimeSchema.ParseGroupVersion(apiResList.GroupVersion)
+		if err != nil {
+			b.log.Debug().Err(err).Str("groupVersion", apiResList.GroupVersion).Msg("failed to parse group version")
+			continue
+		}
+
+		for _, resource := range apiResList.APIResources {
+			// Create a key for group/kind to track preferred version
+			key := fmt.Sprintf("%s/%s", gv.Group, resource.Kind)
+
+			// Store this version as preferred for this group/kind
+			// ServerPreferredResources returns the preferred version for each group
+			b.preferredVersions[key] = gv.Version
+
+			b.log.Debug().
+				Str("group", gv.Group).
+				Str("kind", resource.Kind).
+				Str("preferredVersion", gv.Version).
+				Msg("registered preferred version")
+		}
+	}
+	return b
+}
 
-			gv, err := runtimeSchema.ParseGroupVersion(apiResourceList.GroupVersion)
-			if err != nil {
-				b.err = multierror.Append(b.err, errors.Join(ErrParseGroupVersion, err))
+// WithRelationships adds relationship fields to schemas that have *Ref fields
+// Uses 1-level depth control to prevent circular references and N+1 problems
+func (b *SchemaBuilder) WithRelationships() *SchemaBuilder {
+	// Build kind registry first
+	b.buildKindRegistry()
+
+	// Expand relationships with 1-level depth control
+	b.expandWithSimpleDepthControl()
+
+	return b
+}
+
+// expandWithSimpleDepthControl implements the working 1-level depth control
+func (b *SchemaBuilder) expandWithSimpleDepthControl() {
+	// First pass: identify relation targets
+	relationTargets := make(map[string]bool)
+	for _, schema := range b.schemas {
+		if schema.Properties == nil {
+			continue
+		}
+		for propName := range schema.Properties {
+			if !isRefProperty(propName) {
 				continue
 			}
-			gvk := metav1.GroupVersionKind{
-				Group:   gv.Group,
-				Version: gv.Version,
-				Kind:    apiResource.Kind,
+			baseKind := strings.TrimSuffix(propName, "Ref")
+			candidates := b.findAllCandidatesForKind(baseKind)
+
+			// Mark all candidates as relation targets
+			for _, candidate := range candidates {
+				relationTargets[candidate.SchemaKey] = true
 			}
+		}
+	}
 
-			schema, ok := b.schemas[getOpenAPISchemaKey(gvk)]
-			if !ok {
-				continue
+	b.log.Info().
+		Int("kindRegistrySize", len(b.kindRegistry)).
+		Int("relationTargets", len(relationTargets)).
+		Msg("Starting 1-level relationship expansion")
+
+	// Second pass: expand only non-targets
+	for schemaKey, schema := range b.schemas {
+		if relationTargets[schemaKey] {
+			b.log.Debug().Str("schemaKey", schemaKey).Msg("Skipping relation target (1-level depth control)")
+			continue
+		}
+		b.expandRelationshipsSimple(schema, schemaKey)
+	}
+}
+
+// buildKindRegistry builds a map of kind names to available resource types
+func (b *SchemaBuilder) buildKindRegistry() {
+	for schemaKey, schema := range b.schemas {
+		// Extract GVK from schema
+		if schema.VendorExtensible.Extensions == nil {
+			continue
+		}
+
+		gvksVal, ok := schema.VendorExtensible.Extensions[common.GVKExtensionKey]
+		if !ok {
+			continue
+		}
+
+		jsonBytes, err := json.Marshal(gvksVal)
+		if err != nil {
+			b.err = multierror.Append(b.err, errors.Join(ErrBuildKindRegistry, err))
+			b.log.Debug().Err(err).Str("schemaKey", schemaKey).Msg("failed to marshal GVK")
+			continue
+		}
+
+		var gvks []*GroupVersionKind
+		if err := json.Unmarshal(jsonBytes, &gvks); err != nil {
+			b.err = multierror.Append(b.err, errors.Join(ErrBuildKindRegistry, err))
+			b.log.Debug().Err(err).Str("schemaKey", schemaKey).Msg("failed to unmarshal GVK")
+			continue
+		}
+
+		if len(gvks) != 1 {
+			continue
+		}
+
+		gvk := gvks[0]
+
+		// Add to kind registry with precise GVK key
+		resourceInfo := ResourceInfo{
+			Group:     gvk.Group,
+			Version:   gvk.Version,
+			Kind:      gvk.Kind,
+			SchemaKey: schemaKey,
+		}
+
+		// Index by full GroupVersionKind for precise lookup (no collisions)
+		gvkKey := GroupVersionKind{
+			Group:   gvk.Group,
+			Version: gvk.Version,
+			Kind:    gvk.Kind,
+		}
+		b.kindRegistry[gvkKey] = resourceInfo
+
+	}
+
+	// No sorting needed - each GVK is now uniquely indexed
+	// Check for kinds with multiple resources but no preferred versions
+	b.warnAboutMissingPreferredVersions()
+
+	b.log.Debug().Int("gvkCount", len(b.kindRegistry)).Msg("built kind registry for relationships")
+}
+
+// warnAboutMissingPreferredVersions checks for kinds with multiple resources but no preferred versions
+func (b *SchemaBuilder) warnAboutMissingPreferredVersions() {
+	// Group resources by kind name to find potential conflicts
+	kindGroups := make(map[string][]ResourceInfo)
+
+	for _, resourceInfo := range b.kindRegistry {
+		kindKey := strings.ToLower(resourceInfo.Kind)
+		kindGroups[kindKey] = append(kindGroups[kindKey], resourceInfo)
+	}
+
+	// Check each kind that has multiple resources
+	for kindName, resources := range kindGroups {
+		if len(resources) <= 1 {
+			continue // No conflict possible
+		}
+
+		// Check if any of the resources has a preferred version
+		hasPreferred := false
+		for _, resource := range resources {
+			key := fmt.Sprintf("%s/%s", resource.Group, resource.Kind)
+			if b.preferredVersions[key] == resource.Version {
+				hasPreferred = true
+				break
+			}
+		}
+
+		// Warn if no preferred version found
+		if !hasPreferred {
+			groups := make([]string, 0, len(resources))
+			for _, resource := range resources {
+				groups = append(groups, fmt.Sprintf("%s/%s", resource.Group, resource.Version))
 			}
+			b.log.Warn().
+				Str("kind", kindName).
+				Strs("availableResources", groups).
+				Msg("Multiple resources found for kind with no preferred version - using fallback resolution. Consider setting preferred versions for better API governance.")
+		}
+	}
+}
+
+// expandRelationshipsSimple adds relationship fields for the simple 1-level depth control
+func (b *SchemaBuilder) expandRelationshipsSimple(schema *spec.Schema, schemaKey string) {
+	if schema.Properties == nil {
+		return
+	}
 
-			schema.VendorExtensible.AddExtension(common.CategoriesExtensionKey, apiResource.Categories)
+	for propName := range schema.Properties {
+		if !isRefProperty(propName) {
+			continue
 		}
+
+		baseKind := strings.TrimSuffix(propName, "Ref")
+
+		// Add relationship field using kubectl-style priority resolution
+		b.processReferenceField(schema, schemaKey, propName, baseKind)
 	}
+}
 
-	return b
+// processReferenceField handles individual reference field processing with kubectl-style priority resolution
+func (b *SchemaBuilder) processReferenceField(schema *spec.Schema, schemaKey, propName, baseKind string) {
+	// Find best resource using kubectl-style priority
+	bestResource := b.findBestResourceForKind(baseKind)
+
+	if bestResource == nil {
+		// No candidates found - skip relationship field generation
+		b.log.Debug().
+			Str("kind", baseKind).
+			Str("sourceField", propName).
+			Str("sourceSchema", schemaKey).
+			Msg("No candidates found for kind - skipping relationship field")
+		return
+	}
+
+	// Generate relationship field using the best resource
+	b.addRelationshipField(schema, schemaKey, propName, baseKind, bestResource)
+}
+
+// findBestResourceForKind finds the best resource for a kind using kubectl-style priority resolution
+func (b *SchemaBuilder) findBestResourceForKind(kindName string) *ResourceInfo {
+	candidates := b.findAllCandidatesForKind(kindName)
+
+	if len(candidates) == 0 {
+		return nil
+	}
+
+	if len(candidates) == 1 {
+		return &candidates[0]
+	}
+
+	// Multiple candidates - use kubectl-style priority resolution
+	best := b.selectByKubectlPriority(candidates)
+
+	// Log warning about the conflict for observability
+	groups := make([]string, len(candidates))
+	for i, candidate := range candidates {
+		groups[i] = b.formatGroupVersion(candidate)
+	}
+	b.log.Warn().
+		Str("kind", kindName).
+		Str("selectedGroup", b.formatGroupVersion(best)).
+		Strs("availableGroups", groups).
+		Msg("Multiple API groups provide this kind - selected first by priority (kubectl-style)")
+
+	return &best
+}
+
+// findAllCandidatesForKind finds all resources that match the given kind name
+func (b *SchemaBuilder) findAllCandidatesForKind(kindName string) []ResourceInfo {
+	candidates := make([]ResourceInfo, 0)
+
+	for gvk, resourceInfo := range b.kindRegistry {
+		if strings.EqualFold(gvk.Kind, kindName) {
+			candidates = append(candidates, resourceInfo)
+		}
+	}
+
+	return candidates
+}
+
+// selectByKubectlPriority selects the best resource using kubectl's priority rules
+func (sb *SchemaBuilder) selectByKubectlPriority(candidates []ResourceInfo) ResourceInfo {
+	// Sort candidates by kubectl priority:
+	// 1. Preferred versions first
+	// 2. Core groups (empty group) over extensions
+	// 3. Alphabetical by group name
+	// 4. Alphabetical by version (newer versions typically sort later)
+	slices.SortFunc(candidates, func(a, b ResourceInfo) int {
+		// 1. Check preferred versions first
+		aPreferred := sb.isPreferredVersion(a)
+		bPreferred := sb.isPreferredVersion(b)
+		if aPreferred && !bPreferred {
+			return -1 // a wins
+		}
+		if !aPreferred && bPreferred {
+			return 1 // b wins
+		}
+
+		// 2. Core groups (empty group) beat extension groups
+		aCoreGroup := (a.Group == "")
+		bCoreGroup := (b.Group == "")
+		if aCoreGroup && !bCoreGroup {
+			return -1 // a wins (core group)
+		}
+		if !aCoreGroup && bCoreGroup {
+			return 1 // b wins (core group)
+		}
+
+		// 3. Alphabetical by group name
+		if cmp := strings.Compare(a.Group, b.Group); cmp != 0 {
+			return cmp
+		}
+
+		// 4. Alphabetical by version (this gives deterministic results)
+		return strings.Compare(a.Version, b.Version)
+	})
+
+	return candidates[0] // Return the first (highest priority) candidate
+}
+
+// isPreferredVersion checks if this resource version is marked as preferred
+func (b *SchemaBuilder) isPreferredVersion(resource ResourceInfo) bool {
+	key := fmt.Sprintf("%s/%s", resource.Group, resource.Kind)
+	return b.preferredVersions[key] == resource.Version
+}
+
+// formatGroupVersion formats a resource for display
+func (b *SchemaBuilder) formatGroupVersion(resource ResourceInfo) string {
+	if resource.Group == "" {
+		return fmt.Sprintf("core/%s", resource.Version)
+	}
+	return fmt.Sprintf("%s/%s", resource.Group, resource.Version)
+}
+
+// addRelationshipField adds a relationship field for unambiguous references
+func (b *SchemaBuilder) addRelationshipField(schema *spec.Schema, schemaKey, propName, baseKind string, target *ResourceInfo) {
+	fieldName := strings.ToLower(baseKind)
+	if _, exists := schema.Properties[fieldName]; exists {
+		return
+	}
+
+	// Create proper reference - handle empty group (core) properly
+	var refPath string
+	if target.Group == "" {
+		refPath = fmt.Sprintf("#/definitions/%s.%s", target.Version, target.Kind)
+	} else {
+		refPath = fmt.Sprintf("#/definitions/%s.%s.%s", target.Group, target.Version, target.Kind)
+	}
+	ref := spec.MustCreateRef(refPath)
+	schema.Properties[fieldName] = spec.Schema{SchemaProps: spec.SchemaProps{Ref: ref}}
+
+	b.log.Info().
+		Str("sourceField", propName).
+		Str("targetField", fieldName).
+		Str("targetKind", target.Kind).
+		Str("targetGroup", target.Group).
+		Str("refPath", refPath).
+		Str("sourceSchema", schemaKey).
+		Msg("Added relationship field")
+}
+
+func isRefProperty(name string) bool {
+	if !strings.HasSuffix(name, "Ref") {
+		return false
+	}
+	if name == "Ref" {
+		return false
+	}
+	return true
 }
 
 func (b *SchemaBuilder) Complete() ([]byte, error) {
@@ -180,18 +541,18 @@ func (b *SchemaBuilder) Complete() ([]byte, error) {
 		},
 	})
 	if err != nil {
-		b.err = multierror.Append(b.err, errors.Join(ErrMarshalOpenAPISchema, err))
-		return nil, b.err
+		return nil, errors.Join(ErrMarshalOpenAPISchema, err)
 	}
+
 	v2JSON, err := ConvertJSON(v3JSON)
 	if err != nil {
-		b.err = multierror.Append(b.err, errors.Join(ErrConvertOpenAPISchema, err))
-		return nil, b.err
+		return nil, errors.Join(ErrConvertOpenAPISchema, err)
 	}
 
 	return v2JSON, nil
 }
 
+// getOpenAPISchemaKey creates the key that kubernetes uses in its OpenAPI Definitions
 func getOpenAPISchemaKey(gvk metav1.GroupVersionKind) string {
 	// we need to inverse group to match the runtimeSchema key(io.openmfp.core.v1alpha1.Account)
 	parts := strings.Split(gvk.Group, ".")
diff --git a/listener/pkg/apischema/crd_resolver.go b/listener/pkg/apischema/crd_resolver.go
index bae15b3..b56be55 100644
--- a/listener/pkg/apischema/crd_resolver.go
+++ b/listener/pkg/apischema/crd_resolver.go
@@ -76,7 +76,9 @@ func (cr *CRDResolver) ResolveApiSchema(crd *apiextensionsv1.CustomResourceDefin
 
 	result, err := NewSchemaBuilder(cr.OpenAPIV3(), preferredApiGroups, cr.log).
 		WithScope(cr.RESTMapper).
+		WithPreferredVersions(apiResLists).
 		WithCRDCategories(crd).
+		WithRelationships().
 		Complete()
 
 	if err != nil {
@@ -206,7 +208,9 @@ func (cr *CRDResolver) resolveSchema(dc discovery.DiscoveryInterface, rm meta.RE
 
 	result, err := NewSchemaBuilder(dc.OpenAPIV3(), preferredApiGroups, cr.log).
 		WithScope(rm).
+		WithPreferredVersions(apiResList).
 		WithApiResourceCategories(apiResList).
+		WithRelationships().
 		Complete()
 
 	if err != nil {
diff --git a/listener/pkg/apischema/relationships_test.go b/listener/pkg/apischema/relationships_test.go
new file mode 100644
index 0000000..f6bc5da
--- /dev/null
+++ b/listener/pkg/apischema/relationships_test.go
@@ -0,0 +1,370 @@
+package apischema_test
+
+import (
+	"testing"
+
+	"github.com/openmfp/golang-commons/logger/testlogger"
+	apischema "github.com/openmfp/kubernetes-graphql-gateway/listener/pkg/apischema"
+	apimocks "github.com/openmfp/kubernetes-graphql-gateway/listener/pkg/apischema/mocks"
+	"github.com/stretchr/testify/assert"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/openapi"
+	"k8s.io/kube-openapi/pkg/validation/spec"
+)
+
+// helper constructs a schema with x-kubernetes-group-version-kind
+func schemaWithGVK(group, version, kind string) *spec.Schema {
+	return &spec.Schema{
+		VendorExtensible: spec.VendorExtensible{Extensions: map[string]interface{}{
+			"x-kubernetes-group-version-kind": []map[string]string{{
+				"group":   group,
+				"version": version,
+				"kind":    kind,
+			}},
+		}},
+	}
+}
+
+func Test_with_relationships_adds_single_target_field(t *testing.T) {
+	mock := apimocks.NewMockClient(t)
+	mock.EXPECT().Paths().Return(map[string]openapi.GroupVersion{}, nil)
+	b := apischema.NewSchemaBuilder(mock, nil, testlogger.New().Logger)
+
+	// definitions contain a target kind Role in group g/v1
+	roleKey := "g.v1.Role"
+	roleSchema := schemaWithGVK("g", "v1", "Role")
+
+	// source schema that has roleRef
+	sourceKey := "g2.v1.Binding"
+	sourceSchema := &spec.Schema{SchemaProps: spec.SchemaProps{Properties: map[string]spec.Schema{
+		"roleRef": {SchemaProps: spec.SchemaProps{Type: spec.StringOrArray{"object"}}},
+	}}}
+
+	b.SetSchemas(map[string]*spec.Schema{
+		roleKey:   roleSchema,
+		sourceKey: sourceSchema,
+	})
+
+	b.WithRelationships()
+
+	// Expect that role field was added referencing the Role definition
+	added, ok := b.GetSchemas()[sourceKey].Properties["role"]
+	assert.True(t, ok, "expected relationship field 'role' to be added")
+	assert.True(t, added.Ref.GetURL() != nil, "expected $ref on relationship field")
+	assert.Contains(t, added.Ref.String(), "#/definitions/g.v1.Role")
+}
+
+func Test_kubectl_style_priority_resolution_for_conflicts(t *testing.T) {
+	mock := apimocks.NewMockClient(t)
+	mock.EXPECT().Paths().Return(map[string]openapi.GroupVersion{}, nil)
+	b := apischema.NewSchemaBuilder(mock, nil, testlogger.New().Logger)
+
+	// Two schemas with same Kind different groups - should use kubectl-style priority
+	first := schemaWithGVK("a.example", "v1", "Thing")
+	second := schemaWithGVK("b.example", "v1", "Thing")
+	b.SetSchemas(map[string]*spec.Schema{
+		"a.example.v1.Thing": first,
+		"b.example.v1.Thing": second,
+		"c.other.v1.Other":   schemaWithGVK("c.other", "v1", "Other"),
+	})
+
+	b.WithRelationships() // indirectly builds the registry
+
+	// Add a schema that references thingRef - should use priority resolution
+	sRef := &spec.Schema{SchemaProps: spec.SchemaProps{Properties: map[string]spec.Schema{
+		"thingRef": {SchemaProps: spec.SchemaProps{Type: spec.StringOrArray{"object"}}},
+	}}}
+	b.GetSchemas()["x.v1.HasThing"] = sRef
+
+	b.WithRelationships()
+
+	// Kubectl-style resolution should GENERATE automatic relationship field using priority
+	_, hasAutoField := b.GetSchemas()["x.v1.HasThing"].Properties["thing"]
+	assert.True(t, hasAutoField, "automatic relationship field should be generated using kubectl-style priority")
+
+	// The *Ref field should remain unchanged (backward compatible)
+	thingRefField := b.GetSchemas()["x.v1.HasThing"].Properties["thingRef"]
+	assert.NotContains(t, thingRefField.Required, "apiGroup", "apiGroup should NOT be required - backward compatible")
+	assert.NotContains(t, thingRefField.Required, "kind", "kind should NOT be required - backward compatible")
+}
+
+func Test_kubectl_style_priority_respects_preferred_versions(t *testing.T) {
+	mock := apimocks.NewMockClient(t)
+	mock.EXPECT().Paths().Return(map[string]openapi.GroupVersion{}, nil)
+	b := apischema.NewSchemaBuilder(mock, nil, testlogger.New().Logger)
+
+	// Multiple schemas with same Kind - conflicts exist even with preferred versions
+	childA := schemaWithGVK("a.example", "v1", "Child")
+	childB := schemaWithGVK("b.example", "v1", "Child")
+	childZ := schemaWithGVK("z.last", "v1", "Child") // would be last alphabetically
+
+	b.SetSchemas(map[string]*spec.Schema{
+		"a.example.v1.Child": childA,
+		"b.example.v1.Child": childB,
+		"z.last.v1.Child":    childZ,
+	})
+
+	// Set z.last as preferred (even though it would be last alphabetically)
+	b.WithPreferredVersions([]*metav1.APIResourceList{
+		{
+			GroupVersion: "z.last/v1",
+			APIResources: []metav1.APIResource{
+				{Kind: "Child"},
+			},
+		},
+	})
+
+	b.WithRelationships()
+
+	// Add a parent schema that references childRef
+	parentSchema := &spec.Schema{SchemaProps: spec.SchemaProps{Properties: map[string]spec.Schema{
+		"childRef": {SchemaProps: spec.SchemaProps{Type: spec.StringOrArray{"object"}}},
+	}}}
+	b.GetSchemas()["x.v1.Parent"] = parentSchema
+
+	b.WithRelationships()
+
+	// Kubectl-style resolution should use preferred version to generate relationship field
+	_, hasAutoField := b.GetSchemas()["x.v1.Parent"].Properties["child"]
+	assert.True(t, hasAutoField, "automatic relationship field should be generated using preferred version priority")
+
+	// The *Ref field should remain unchanged (backward compatible)
+	childRefField := b.GetSchemas()["x.v1.Parent"].Properties["childRef"]
+	assert.NotContains(t, childRefField.Required, "apiGroup", "apiGroup should NOT be required - backward compatible")
+	assert.NotContains(t, childRefField.Required, "kind", "kind should NOT be required - backward compatible")
+}
+
+func Test_depth_control_prevents_deep_nesting(t *testing.T) {
+	mock := apimocks.NewMockClient(t)
+	mock.EXPECT().Paths().Return(map[string]openapi.GroupVersion{}, nil)
+	b := apischema.NewSchemaBuilder(mock, nil, testlogger.New().Logger)
+
+	// Create a chain: Root -> Pod -> Service
+	// Only Root should get relationship fields, Pod and Service should be marked as targets
+	rootSchema := schemaWithGVK("example.com", "v1", "Root")
+	rootSchema.Properties = map[string]spec.Schema{
+		"podRef": {SchemaProps: spec.SchemaProps{Type: spec.StringOrArray{"object"}}},
+	}
+
+	podSchema := schemaWithGVK("", "v1", "Pod") // Core group
+	podSchema.Properties = map[string]spec.Schema{
+		"serviceRef": {SchemaProps: spec.SchemaProps{Type: spec.StringOrArray{"object"}}},
+	}
+
+	serviceSchema := schemaWithGVK("", "v1", "Service") // Core group
+
+	b.SetSchemas(map[string]*spec.Schema{
+		"example.com.v1.Root": rootSchema,
+		".v1.Pod":             podSchema,
+		".v1.Service":         serviceSchema,
+	})
+
+	// Verify default depth is 1
+	b.WithRelationships()
+
+	schemas := b.GetSchemas()
+
+	// Root should get 'pod' relationship field (depth 0 -> 1)
+	_, hasPodField := schemas["example.com.v1.Root"].Properties["pod"]
+	assert.True(t, hasPodField, "Root should get pod relationship field")
+
+	// Pod should NOT get 'service' relationship field (would be depth 1 -> 2, exceeds limit)
+	_, hasServiceField := schemas[".v1.Pod"].Properties["service"]
+	assert.False(t, hasServiceField, "Pod should NOT get service relationship field due to depth limit")
+
+	// Service should not have any relationship fields added
+	originalServiceProps := len(serviceSchema.Properties)
+	currentServiceProps := len(schemas[".v1.Service"].Properties)
+	assert.Equal(t, originalServiceProps, currentServiceProps, "Service should not have relationship fields added")
+}
+
+func Test_single_level_prevents_circular_relationships(t *testing.T) {
+	mock := apimocks.NewMockClient(t)
+	mock.EXPECT().Paths().Return(map[string]openapi.GroupVersion{}, nil)
+	b := apischema.NewSchemaBuilder(mock, nil, testlogger.New().Logger)
+
+	// Create circular reference: A -> B, B -> A
+	aSchema := schemaWithGVK("example.com", "v1", "A")
+	aSchema.Properties = map[string]spec.Schema{
+		"bRef": {SchemaProps: spec.SchemaProps{Type: spec.StringOrArray{"object"}}},
+	}
+
+	bSchema := schemaWithGVK("example.com", "v1", "B")
+	bSchema.Properties = map[string]spec.Schema{
+		"aRef": {SchemaProps: spec.SchemaProps{Type: spec.StringOrArray{"object"}}},
+	}
+
+	b.SetSchemas(map[string]*spec.Schema{
+		"example.com.v1.A": aSchema,
+		"example.com.v1.B": bSchema,
+	})
+
+	b.WithRelationships()
+
+	schemas := b.GetSchemas()
+
+	// With 1-level depth control, both A and B are marked as relation targets
+	// So neither should get automatic relationship fields
+	_, hasAField := schemas["example.com.v1.A"].Properties["b"]
+	_, hasBField := schemas["example.com.v1.B"].Properties["a"]
+
+	// At least one should not have the field to prevent infinite circular expansion
+	circularPrevented := !hasAField || !hasBField
+	assert.True(t, circularPrevented, "Circular relationships should be prevented by depth control")
+}
+
+func Test_depth_control_with_multiple_chains(t *testing.T) {
+	mock := apimocks.NewMockClient(t)
+	mock.EXPECT().Paths().Return(map[string]openapi.GroupVersion{}, nil)
+	b := apischema.NewSchemaBuilder(mock, nil, testlogger.New().Logger)
+
+	// Multiple chains: Chain1 (Root1 -> Pod), Chain2 (Root2 -> Service)
+	root1Schema := schemaWithGVK("example.com", "v1", "Root1")
+	root1Schema.Properties = map[string]spec.Schema{
+		"podRef": {SchemaProps: spec.SchemaProps{Type: spec.StringOrArray{"object"}}},
+	}
+
+	root2Schema := schemaWithGVK("example.com", "v1", "Root2")
+	root2Schema.Properties = map[string]spec.Schema{
+		"serviceRef": {SchemaProps: spec.SchemaProps{Type: spec.StringOrArray{"object"}}},
+	}
+
+	podSchema := schemaWithGVK("", "v1", "Pod")
+	serviceSchema := schemaWithGVK("", "v1", "Service")
+
+	b.SetSchemas(map[string]*spec.Schema{
+		"example.com.v1.Root1": root1Schema,
+		"example.com.v1.Root2": root2Schema,
+		".v1.Pod":              podSchema,
+		".v1.Service":          serviceSchema,
+	})
+
+	b.WithRelationships()
+
+	schemas := b.GetSchemas()
+
+	// Both roots should be able to reference their targets (no conflicts between chains)
+	_, hasPodField := schemas["example.com.v1.Root1"].Properties["pod"]
+	_, hasServiceField := schemas["example.com.v1.Root2"].Properties["service"]
+
+	assert.True(t, hasPodField, "Root1 should get pod relationship field")
+	assert.True(t, hasServiceField, "Root2 should get service relationship field")
+
+	// Targets (Pod, Service) should not get any additional relationship fields
+	assert.Empty(t, schemas[".v1.Pod"].Properties, "Pod should not have relationship fields (is a target)")
+	assert.Empty(t, schemas[".v1.Service"].Properties, "Service should not have relationship fields (is a target)")
+}
+
+func Test_same_kind_different_groups_with_explicit_disambiguation(t *testing.T) {
+	mock := apimocks.NewMockClient(t)
+	mock.EXPECT().Paths().Return(map[string]openapi.GroupVersion{}, nil)
+	b := apischema.NewSchemaBuilder(mock, nil, testlogger.New().Logger)
+
+	// Create two different groups providing the same "Database" kind
+	mysqlDB := schemaWithGVK("mysql.example.com", "v1", "Database")
+	postgresDB := schemaWithGVK("postgres.example.com", "v1", "Database")
+
+	// Parent schema that wants to reference one of the databases
+	appSchema := schemaWithGVK("apps.example.com", "v1", "Application")
+	appSchema.Properties = map[string]spec.Schema{
+		"databaseRef": {SchemaProps: spec.SchemaProps{Type: spec.StringOrArray{"object"}}},
+	}
+
+	b.SetSchemas(map[string]*spec.Schema{
+		"mysql.example.com.v1.Database":    mysqlDB,
+		"postgres.example.com.v1.Database": postgresDB,
+		"apps.example.com.v1.Application":  appSchema,
+	})
+
+	b.WithRelationships()
+
+	schemas := b.GetSchemas()
+
+	// Verify kubectl-style resolution was applied
+	_, hasAutoField := schemas["apps.example.com.v1.Application"].Properties["database"]
+	assert.True(t, hasAutoField, "automatic relationship field should be generated using kubectl-style priority")
+
+	// Verify the databaseRef field remains unchanged (backward compatible)
+	dbRefField := schemas["apps.example.com.v1.Application"].Properties["databaseRef"]
+	assert.NotContains(t, dbRefField.Required, "apiGroup", "apiGroup should NOT be required - backward compatible")
+	assert.NotContains(t, dbRefField.Required, "kind", "kind should NOT be required - backward compatible")
+}
+
+func Test_same_kind_different_groups_kubernetes_core_vs_custom(t *testing.T) {
+	mock := apimocks.NewMockClient(t)
+	mock.EXPECT().Paths().Return(map[string]openapi.GroupVersion{}, nil)
+	b := apischema.NewSchemaBuilder(mock, nil, testlogger.New().Logger)
+
+	// Simulate core Kubernetes Service vs custom Service
+	coreService := schemaWithGVK("", "v1", "Service") // Core group (empty)
+	customService := schemaWithGVK("custom.example.com", "v1", "Service")
+
+	// Parent that references "Service" - which one?
+	parentSchema := schemaWithGVK("example.com", "v1", "Parent")
+	parentSchema.Properties = map[string]spec.Schema{
+		"serviceRef": {SchemaProps: spec.SchemaProps{Type: spec.StringOrArray{"object"}}},
+	}
+
+	b.SetSchemas(map[string]*spec.Schema{
+		".v1.Service":                   coreService,
+		"custom.example.com.v1.Service": customService,
+		"example.com.v1.Parent":         parentSchema,
+	})
+
+	b.WithRelationships()
+
+	schemas := b.GetSchemas()
+
+	// Even with core vs custom, should still require disambiguation
+	_, hasAutoField := schemas["example.com.v1.Parent"].Properties["service"]
+	assert.True(t, hasAutoField, "automatic relationship field should be generated using kubectl-style priority (core wins)")
+
+	// The serviceRef field should remain unchanged (backward compatible)
+	serviceRefField := schemas["example.com.v1.Parent"].Properties["serviceRef"]
+	assert.NotContains(t, serviceRefField.Required, "apiGroup", "apiGroup should NOT be required - backward compatible")
+	assert.NotContains(t, serviceRefField.Required, "kind", "kind should NOT be required - backward compatible")
+}
+
+func Test_same_kind_different_groups_with_preferred_version_still_conflicts(t *testing.T) {
+	mock := apimocks.NewMockClient(t)
+	mock.EXPECT().Paths().Return(map[string]openapi.GroupVersion{}, nil)
+	b := apischema.NewSchemaBuilder(mock, nil, testlogger.New().Logger)
+
+	// Multiple "Storage" providers with preferred version set
+	s3Storage := schemaWithGVK("aws.example.com", "v1", "Storage")
+	gcsStorage := schemaWithGVK("gcp.example.com", "v1", "Storage")
+	azureStorage := schemaWithGVK("azure.example.com", "v1", "Storage")
+
+	b.SetSchemas(map[string]*spec.Schema{
+		"aws.example.com.v1.Storage":   s3Storage,
+		"gcp.example.com.v1.Storage":   gcsStorage,
+		"azure.example.com.v1.Storage": azureStorage,
+	})
+
+	// Set preferred version for one of them
+	b.WithPreferredVersions([]*metav1.APIResourceList{
+		{
+			GroupVersion: "aws.example.com/v1",
+			APIResources: []metav1.APIResource{{Kind: "Storage"}},
+		},
+	})
+
+	// Parent that wants to reference storage
+	appSchema := schemaWithGVK("apps.example.com", "v1", "BackupApp")
+	appSchema.Properties = map[string]spec.Schema{
+		"storageRef": {SchemaProps: spec.SchemaProps{Type: spec.StringOrArray{"object"}}},
+	}
+	b.GetSchemas()["apps.example.com.v1.BackupApp"] = appSchema
+
+	b.WithRelationships()
+
+	schemas := b.GetSchemas()
+
+	// Kubectl-style resolution should use preferred version and generate relationship field
+	_, hasAutoField := schemas["apps.example.com.v1.BackupApp"].Properties["storage"]
+	assert.True(t, hasAutoField, "automatic relationship field should be generated using preferred version priority")
+
+	// The storageRef field should remain unchanged (backward compatible)
+	storageRefField := schemas["apps.example.com.v1.BackupApp"].Properties["storageRef"]
+	assert.NotContains(t, storageRefField.Required, "apiGroup", "apiGroup should NOT be required - backward compatible")
+}
diff --git a/listener/reconciler/kcp/virtual_workspace_test.go b/listener/reconciler/kcp/virtual_workspace_test.go
index 03373ce..babf738 100644
--- a/listener/reconciler/kcp/virtual_workspace_test.go
+++ b/listener/reconciler/kcp/virtual_workspace_test.go
@@ -695,8 +695,13 @@ func TestVirtualWorkspaceReconciler_ProcessVirtualWorkspace(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			// Set up test environment where KUBECONFIG is not available
 			oldKubeconfig := os.Getenv("KUBECONFIG")
-			defer os.Setenv("KUBECONFIG", oldKubeconfig)
+			oldHome := os.Getenv("HOME")
+			defer func() {
+				os.Setenv("KUBECONFIG", oldKubeconfig)
+				os.Setenv("HOME", oldHome)
+			}()
 			os.Unsetenv("KUBECONFIG")
+			os.Setenv("HOME", "/nonexistent") // Force metadata injection to fail consistently
 
 			appCfg := config.Config{}
 			appCfg.Url.VirtualWorkspacePrefix = "virtual-workspace"
diff --git a/tests/gateway_test/helpers_rbac_authorization_k8s_io_test.go b/tests/gateway_test/helpers_rbac_authorization_k8s_io_test.go
index 5ecd3f2..33843c7 100644
--- a/tests/gateway_test/helpers_rbac_authorization_k8s_io_test.go
+++ b/tests/gateway_test/helpers_rbac_authorization_k8s_io_test.go
@@ -1,13 +1,30 @@
 package gateway_test
 
 type RbacAuthorizationK8sIO struct {
-	ClusterRole *ClusterRole `json:"ClusterRole,omitempty"`
+	ClusterRole        *ClusterRole        `json:"ClusterRole,omitempty"`
+	ClusterRoleBinding *ClusterRoleBinding `json:"ClusterRoleBinding,omitempty"`
 }
 
 type ClusterRole struct {
 	Metadata metadata `json:"metadata"`
 }
 
+type ClusterRoleBinding struct {
+	Metadata metadata `json:"metadata"`
+	RoleRef  roleRef  `json:"roleRef"`
+}
+
+type roleRef struct {
+	Name     string `json:"name"`
+	Kind     string `json:"kind"`
+	APIGroup string `json:"apiGroup"`
+	Role     crMeta `json:"role"`
+}
+
+type crMeta struct {
+	Metadata metadata `json:"metadata"`
+}
+
 func CreateClusterRoleMutation() string {
 	return `mutation {
 			  rbac_authorization_k8s_io {
diff --git a/tests/gateway_test/relation_rbac_test.go b/tests/gateway_test/relation_rbac_test.go
new file mode 100644
index 0000000..967af81
--- /dev/null
+++ b/tests/gateway_test/relation_rbac_test.go
@@ -0,0 +1,113 @@
+package gateway_test
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"path/filepath"
+
+	"github.com/stretchr/testify/require"
+)
+
+// Test_relation_clusterrolebinding_role_ref mirrors pod test style: creates schema file per workspace,
+// creates a ClusterRole and ClusterRoleBinding via GraphQL, then queries roleRef.role to ensure relation resolution.
+func (suite *CommonTestSuite) Test_relation_clusterrolebinding_role_ref() {
+	workspaceName := "relationsWorkspace"
+
+	require.NoError(suite.T(), suite.writeToFileWithClusterMetadata(
+		filepath.Join("testdata", "kubernetes"),
+		filepath.Join(suite.appCfg.OpenApiDefinitionsPath, workspaceName),
+	))
+
+	url := fmt.Sprintf("%s/%s/graphql", suite.server.URL, workspaceName)
+
+	// Create ClusterRole
+	statusCode, body := suite.doRawGraphQL(url, createClusterRoleForRelationMutation())
+	require.Equal(suite.T(), http.StatusOK, statusCode)
+	require.Nil(suite.T(), body["errors"])
+
+	// Create ClusterRoleBinding referencing the ClusterRole
+	statusCode, body = suite.doRawGraphQL(url, createClusterRoleBindingForRelationMutation())
+	require.Equal(suite.T(), http.StatusOK, statusCode)
+	require.Nil(suite.T(), body["errors"])
+
+	// Query ClusterRoleBinding and expand roleRef.role
+	statusCode, body = suite.doRawGraphQL(url, getClusterRoleBindingWithRoleQuery())
+	require.Equal(suite.T(), http.StatusOK, statusCode)
+	require.Nil(suite.T(), body["errors"])
+
+	// Extract nested role name from generic map
+	data, _ := body["data"].(map[string]interface{})
+	rbac, _ := data["rbac_authorization_k8s_io"].(map[string]interface{})
+	crb, _ := rbac["ClusterRoleBinding"].(map[string]interface{})
+	roleRef, _ := crb["roleRef"].(map[string]interface{})
+	role, _ := roleRef["role"].(map[string]interface{})
+	metadata, _ := role["metadata"].(map[string]interface{})
+	name, _ := metadata["name"].(string)
+	require.Equal(suite.T(), "test-cluster-role-rel", name)
+}
+
+// local helper mirroring helpers_test.go but returning generic body
+func (suite *CommonTestSuite) doRawGraphQL(url, query string) (int, map[string]interface{}) {
+	reqBody := map[string]string{"query": query}
+	buf, _ := json.Marshal(reqBody)
+	req, _ := http.NewRequest("POST", url, bytes.NewReader(buf))
+	req.Header.Set("Content-Type", "application/json")
+	// add auth token used by suite
+	if suite.staticToken != "" {
+		req.Header.Set("Authorization", "Bearer "+suite.staticToken)
+	}
+	resp, err := http.DefaultClient.Do(req)
+	require.NoError(suite.T(), err)
+	defer resp.Body.Close()
+	var body map[string]interface{}
+	dec := json.NewDecoder(resp.Body)
+	require.NoError(suite.T(), dec.Decode(&body))
+	return resp.StatusCode, body
+}
+
+// GraphQL payloads
+func createClusterRoleForRelationMutation() string {
+	return `mutation {
+  rbac_authorization_k8s_io {
+    createClusterRole(
+      object: {
+        metadata: { name: "test-cluster-role-rel" }
+        rules: [{ apiGroups:[""], resources:["pods"], verbs:["get","list"] }]
+      }
+    ) { metadata { name } }
+  }
+}`
+}
+
+func createClusterRoleBindingForRelationMutation() string {
+	return `mutation {
+  rbac_authorization_k8s_io {
+    createClusterRoleBinding(
+      object: {
+        metadata: { name: "test-crb-rel" }
+        roleRef: {
+          apiGroup: "rbac.authorization.k8s.io"
+          kind: "ClusterRole"
+          name: "test-cluster-role-rel"
+        }
+        subjects: []
+      }
+    ) { metadata { name } }
+  }
+}`
+}
+
+func getClusterRoleBindingWithRoleQuery() string {
+	return `{
+  rbac_authorization_k8s_io {
+    ClusterRoleBinding(name: "test-crb-rel") {
+      roleRef {
+        name kind apiGroup
+        role { metadata { name } }
+      }
+    }
+  }
+}`
+}