diff --git a/agent/grpc-external/services/resource/write.go b/agent/grpc-external/services/resource/write.go
index 121c7ce39c..ad17e61c51 100644
--- a/agent/grpc-external/services/resource/write.go
+++ b/agent/grpc-external/services/resource/write.go
@@ -119,6 +119,31 @@ func (s *Server) Write(ctx context.Context, req *pbresource.WriteRequest) (*pbre
 				return errUseWriteStatus
 			}
 
+			// Generally, we expect resources with owners to be created by controllers,
+			// and they should provide the Uid. In cases where no Uid is given (e.g. the
+			// owner is specified in the resource HCL) we'll look up whatever the current
+			// Uid is and use that.
+			//
+			// An important note on consistency:
+			//
+			// We read the owner with StrongConsistency here to reduce the likelihood of
+			// creating a resource pointing to the wrong "incarnation" of the owner in
+			// cases where the owner is deleted and re-created in quick succession.
+			//
+			// That said, there is still a chance that the owner has been deleted by the
+			// time we write this resource. This is not a relational database and we do
+			// not support ACID transactions or real foreign key constraints.
+			if input.Owner != nil && input.Owner.Uid == "" {
+				owner, err := s.Backend.Read(ctx, storage.StrongConsistency, input.Owner)
+				switch {
+				case errors.Is(err, storage.ErrNotFound):
+					return status.Error(codes.InvalidArgument, "resource.owner does not exist")
+				case err != nil:
+					return status.Errorf(codes.Internal, "failed to resolve owner: %v", err)
+				}
+				input.Owner = owner.Id
+			}
+
 			// TODO(spatel): Revisit owner<->resource tenancy rules post-1.16
 
 		// Update path.
diff --git a/agent/grpc-external/services/resource/write_test.go b/agent/grpc-external/services/resource/write_test.go
index f8620f4a34..b2960a9243 100644
--- a/agent/grpc-external/services/resource/write_test.go
+++ b/agent/grpc-external/services/resource/write_test.go
@@ -491,6 +491,57 @@ func TestWrite_Owner_Immutable(t *testing.T) {
 	require.ErrorContains(t, err, "owner cannot be changed")
 }
 
+func TestWrite_Owner_Uid(t *testing.T) {
+	server := testServer(t)
+	client := testClient(t, server)
+
+	demo.RegisterTypes(server.Registry)
+
+	t.Run("uid given", func(t *testing.T) {
+		artist, err := demo.GenerateV2Artist()
+		require.NoError(t, err)
+
+		album, err := demo.GenerateV2Album(artist.Id)
+		require.NoError(t, err)
+		album.Owner.Uid = ulid.Make().String()
+
+		_, err = client.Write(testContext(t), &pbresource.WriteRequest{Resource: album})
+		require.NoError(t, err)
+	})
+
+	t.Run("no uid - owner not found", func(t *testing.T) {
+		artist, err := demo.GenerateV2Artist()
+		require.NoError(t, err)
+
+		album, err := demo.GenerateV2Album(artist.Id)
+		require.NoError(t, err)
+
+		_, err = client.Write(testContext(t), &pbresource.WriteRequest{Resource: album})
+		require.Error(t, err)
+		require.Equal(t, codes.InvalidArgument.String(), status.Code(err).String())
+	})
+
+	t.Run("no uid - automatically resolved", func(t *testing.T) {
+		artist, err := demo.GenerateV2Artist()
+		require.NoError(t, err)
+
+		rsp1, err := client.Write(testContext(t), &pbresource.WriteRequest{Resource: artist})
+		require.NoError(t, err)
+		artist = rsp1.Resource
+
+		album, err := demo.GenerateV2Album(clone(artist.Id))
+		require.NoError(t, err)
+
+		// Blank out the owner Uid to check it gets automatically filled in.
+		album.Owner.Uid = ""
+
+		rsp2, err := client.Write(testContext(t), &pbresource.WriteRequest{Resource: album})
+		require.NoError(t, err)
+		require.NotEmpty(t, rsp2.Resource.Owner.Uid)
+		require.Equal(t, artist.Id.Uid, rsp2.Resource.Owner.Uid)
+	})
+}
+
 type blockOnceBackend struct {
 	storage.Backend