diff --git a/private/buf/bufctl/controller.go b/private/buf/bufctl/controller.go
index 1cb04b8e30..d4543c502e 100644
--- a/private/buf/bufctl/controller.go
+++ b/private/buf/bufctl/controller.go
@@ -29,6 +29,7 @@ import (
 	"github.com/bufbuild/buf/private/buf/bufwkt/bufwktstore"
 	"github.com/bufbuild/buf/private/buf/bufworkspace"
 	"github.com/bufbuild/buf/private/bufpkg/bufanalysis"
+	"github.com/bufbuild/buf/private/bufpkg/bufcheck"
 	"github.com/bufbuild/buf/private/bufpkg/bufconfig"
 	"github.com/bufbuild/buf/private/bufpkg/bufimage"
 	"github.com/bufbuild/buf/private/bufpkg/bufimage/bufimageutil"
@@ -48,6 +49,7 @@ import (
 	"github.com/bufbuild/buf/private/pkg/slicesext"
 	"github.com/bufbuild/buf/private/pkg/storage/storageos"
 	"github.com/bufbuild/buf/private/pkg/syserror"
+	"github.com/bufbuild/buf/private/pkg/wasm"
 	"github.com/bufbuild/protovalidate-go"
 	"google.golang.org/protobuf/proto"
 )
@@ -128,6 +130,17 @@ type Controller interface {
 		defaultMessageEncoding buffetch.MessageEncoding,
 		options ...FunctionOption,
 	) error
+	// GetCheckRunnerProvider gets a CheckRunnerProvider for the given input.
+	//
+	// The returned RunnerProvider will be able to run lint and breaking checks
+	// using the PluginConfigs from the input. The input provided will resolve
+	// the PluginKeys from the related buf.lock file.
+	GetCheckRunnerProvider(
+		ctx context.Context,
+		input string,
+		wasmRuntime wasm.Runtime,
+		options ...FunctionOption,
+	) (bufcheck.RunnerProvider, error)
 }
 
 func NewController(
@@ -243,6 +256,7 @@ func newController(
 		graphProvider,
 		moduleDataProvider,
 		commitProvider,
+		pluginKeyProvider,
 	)
 	controller.workspaceDepManagerProvider = bufworkspace.NewWorkspaceDepManagerProvider(
 		logger,
@@ -706,6 +720,32 @@ func (c *controller) PutMessage(
 	return errors.Join(err, writeCloser.Close())
 }
 
+func (c *controller) GetCheckRunnerProvider(
+	ctx context.Context,
+	input string,
+	wasmRuntime wasm.Runtime,
+	options ...FunctionOption,
+) (_ bufcheck.RunnerProvider, retErr error) {
+	defer c.handleFileAnnotationSetRetError(&retErr)
+	functionOptions := newFunctionOptions(c)
+	for _, option := range options {
+		option(functionOptions)
+	}
+	ref, err := c.buffetchRefParser.GetRef(ctx, input)
+	if err != nil {
+		return nil, err
+	}
+	pluginKeyProvider, err := c.getPluginKeyProviderForRef(ctx, ref, functionOptions)
+	if err != nil {
+		return nil, err
+	}
+	return bufcheck.NewLocalRunnerProvider(
+		wasmRuntime,
+		pluginKeyProvider,
+		c.pluginDataProvider,
+	), nil
+}
+
 func (c *controller) getImage(
 	ctx context.Context,
 	input string,
@@ -1159,6 +1199,81 @@ Declare %q in the deps key in your buf.yaml.`,
 	return nil
 }
 
+// getPluginKeyProviderForRef create a new PluginKeyProvider for the Ref.
+//
+// Remote plugins Refs are resolved to PluginKeys from the workspace buf.lock file.
+// If the Ref is a MessageRef, we use the current directory buf.lock file.
+func (c *controller) getPluginKeyProviderForRef(
+	ctx context.Context,
+	ref buffetch.Ref,
+	functionOptions *functionOptions,
+) (_ bufplugin.PluginKeyProvider, retErr error) {
+	switch t := ref.(type) {
+	case buffetch.ProtoFileRef:
+		workspace, err := c.getWorkspaceForProtoFileRef(ctx, t, functionOptions)
+		if err != nil {
+			return nil, err
+		}
+		return bufplugin.NewStaticPluginKeyProvider(workspace.RemotePluginKeys())
+	case buffetch.SourceRef:
+		workspace, err := c.getWorkspaceForSourceRef(ctx, t, functionOptions)
+		if err != nil {
+			return nil, err
+		}
+		return bufplugin.NewStaticPluginKeyProvider(workspace.RemotePluginKeys())
+	case buffetch.ModuleRef:
+		workspace, err := c.getWorkspaceForModuleRef(ctx, t, functionOptions)
+		if err != nil {
+			return nil, err
+		}
+		return bufplugin.NewStaticPluginKeyProvider(workspace.RemotePluginKeys())
+	case buffetch.MessageRef:
+		bucket, err := c.storageosProvider.NewReadWriteBucket(
+			".",
+			storageos.ReadWriteBucketWithSymlinksIfSupported(),
+		)
+		if err != nil {
+			return nil, err
+		}
+		bufYAMLFile, err := bufconfig.GetBufYAMLFileForPrefixOrOverride(
+			ctx,
+			bucket,
+			".",
+			functionOptions.configOverride,
+		)
+		if err != nil {
+			if !errors.Is(err, fs.ErrNotExist) {
+				return nil, err
+			}
+			// We did not find a buf.yaml in our current directory,
+			// and there was no config override.
+			return bufplugin.NopPluginKeyProvider, nil
+		}
+		var pluginKeys []bufplugin.PluginKey
+		if bufYAMLFile.FileVersion() == bufconfig.FileVersionV2 {
+			bufLockFile, err := bufconfig.GetBufLockFileForPrefix(
+				ctx,
+				bucket,
+				// buf.lock files live next to the buf.yaml
+				".",
+			)
+			if err != nil {
+				if !errors.Is(err, fs.ErrNotExist) {
+					return nil, err
+				}
+				// We did not find a buf.lock in our current directory.
+				// Remote plugins are not available.
+				return bufplugin.NopPluginKeyProvider, nil
+			}
+			pluginKeys = bufLockFile.RemotePluginKeys()
+		}
+		return bufplugin.NewStaticPluginKeyProvider(pluginKeys)
+	default:
+		// This is a system error.
+		return nil, syserror.Newf("invalid Ref: %T", ref)
+	}
+}
+
 // handleFileAnnotationSetError will attempt to handle the error as a FileAnnotationSet, and if so, print
 // the FileAnnotationSet to the writer with the given error format while returning ErrFileAnnotation.
 //
diff --git a/private/buf/buflsp/buflsp.go b/private/buf/buflsp/buflsp.go
index 536cf71d43..2d440f64f7 100644
--- a/private/buf/buflsp/buflsp.go
+++ b/private/buf/buflsp/buflsp.go
@@ -25,11 +25,11 @@ import (
 	"sync/atomic"
 
 	"github.com/bufbuild/buf/private/buf/bufctl"
-	"github.com/bufbuild/buf/private/bufpkg/bufcheck"
 	"github.com/bufbuild/buf/private/pkg/app/appext"
 	"github.com/bufbuild/buf/private/pkg/slogext"
 	"github.com/bufbuild/buf/private/pkg/storage"
 	"github.com/bufbuild/buf/private/pkg/storage/storageos"
+	"github.com/bufbuild/buf/private/pkg/wasm"
 	"go.lsp.dev/jsonrpc2"
 	"go.lsp.dev/protocol"
 	"go.uber.org/zap"
@@ -43,7 +43,7 @@ func Serve(
 	wktBucket storage.ReadBucket,
 	container appext.Container,
 	controller bufctl.Controller,
-	checkClient bufcheck.Client,
+	wasmRuntime wasm.Runtime,
 	stream jsonrpc2.Stream,
 ) (jsonrpc2.Conn, error) {
 	// The LSP protocol deals with absolute filesystem paths. This requires us to
@@ -68,7 +68,7 @@ func Serve(
 		container:   container,
 		logger:      container.Logger(),
 		controller:  controller,
-		checkClient: checkClient,
+		wasmRuntime: wasmRuntime,
 		rootBucket:  bucket,
 		wktBucket:   wktBucket,
 	}
@@ -96,7 +96,7 @@ type lsp struct {
 
 	logger      *slog.Logger
 	controller  bufctl.Controller
-	checkClient bufcheck.Client
+	wasmRuntime wasm.Runtime
 	rootBucket  storage.ReadBucket
 	fileManager *fileManager
 
diff --git a/private/buf/buflsp/file.go b/private/buf/buflsp/file.go
index 8bad05ec37..c655088edf 100644
--- a/private/buf/buflsp/file.go
+++ b/private/buf/buflsp/file.go
@@ -59,8 +59,9 @@ type file struct {
 	version int32
 	hasText bool // Whether this file has ever had text read into it.
 
-	workspace bufworkspace.Workspace
-	module    bufmodule.Module
+	workspace   bufworkspace.Workspace
+	module      bufmodule.Module
+	checkClient bufcheck.Client
 
 	againstStrategy againstStrategy
 	againstGitRef   string
@@ -371,6 +372,24 @@ func (f *file) FindModule(ctx context.Context) {
 		return
 	}
 
+	// Get the check runner provider for this file. The client is scoped to
+	// the input Buf lock file, so we need to get the check runner provider
+	// for the workspace that contains this file.
+	checkRunnerProvider, err := f.lsp.controller.GetCheckRunnerProvider(ctx, f.uri.Filename(), f.lsp.wasmRuntime)
+	if err != nil {
+		f.lsp.logger.Warn("could not get check runner provider", slogext.ErrorAttr(err))
+		return
+	}
+	checkClient, err := bufcheck.NewClient(
+		f.lsp.logger,
+		checkRunnerProvider,
+		bufcheck.ClientWithStderr(f.lsp.container.Stderr()),
+	)
+	if err != nil {
+		f.lsp.logger.Warn("could not create check client", slogext.ErrorAttr(err))
+		return
+	}
+
 	// Figure out which module this file belongs to.
 	var module bufmodule.Module
 	for _, mod := range workspace.Modules() {
@@ -398,6 +417,7 @@ func (f *file) FindModule(ctx context.Context) {
 
 	f.workspace = workspace
 	f.module = module
+	f.checkClient = checkClient
 }
 
 // IndexImports finds URIs for all of the files imported by this file.
@@ -641,9 +661,14 @@ func (f *file) RunLints(ctx context.Context) bool {
 		f.lsp.logger.Warn(fmt.Sprintf("could not find image for %q", f.uri))
 		return false
 	}
-
+	if f.checkClient == nil {
+		f.lsp.logger.Warn(fmt.Sprintf("could not find check client for %q", f.uri))
+		return false
+	}
 	f.lsp.logger.Debug(fmt.Sprintf("running lint for %q in %v", f.uri, f.module.FullName()))
-	return f.appendLintErrors("buf lint", f.lsp.checkClient.Lint(
+
+	f.lsp.logger.Debug(fmt.Sprintf("running lint for %q in %v %d plugin configs", f.uri, f.module.FullName(), len(f.workspace.PluginConfigs())))
+	return f.appendLintErrors("buf lint", f.checkClient.Lint(
 		ctx,
 		f.workspace.GetLintConfigForOpaqueID(f.module.OpaqueID()),
 		f.image,
@@ -664,9 +689,13 @@ func (f *file) RunBreaking(ctx context.Context) bool {
 		f.lsp.logger.Warn(fmt.Sprintf("could not find --against image for %q", f.uri))
 		return false
 	}
+	if f.checkClient == nil {
+		f.lsp.logger.Warn(fmt.Sprintf("could not find check client for %q", f.uri))
+		return false
+	}
 
 	f.lsp.logger.Debug(fmt.Sprintf("running breaking for %q in %v", f.uri, f.module.FullName()))
-	return f.appendLintErrors("buf breaking", f.lsp.checkClient.Breaking(
+	return f.appendLintErrors("buf breaking", f.checkClient.Breaking(
 		ctx,
 		f.workspace.GetBreakingConfigForOpaqueID(f.module.OpaqueID()),
 		f.image,
diff --git a/private/buf/bufmigrate/migrator.go b/private/buf/bufmigrate/migrator.go
index 0a17a2b33f..cf69bb2960 100644
--- a/private/buf/bufmigrate/migrator.go
+++ b/private/buf/bufmigrate/migrator.go
@@ -29,6 +29,7 @@ import (
 	"github.com/bufbuild/buf/private/bufpkg/bufconfig"
 	"github.com/bufbuild/buf/private/bufpkg/bufmodule"
 	"github.com/bufbuild/buf/private/bufpkg/bufparse"
+	"github.com/bufbuild/buf/private/bufpkg/bufplugin"
 	"github.com/bufbuild/buf/private/pkg/normalpath"
 	"github.com/bufbuild/buf/private/pkg/slicesext"
 	"github.com/bufbuild/buf/private/pkg/storage"
@@ -695,7 +696,11 @@ func equivalentCheckConfigInV2(
 ) (bufconfig.CheckConfig, error) {
 	// No need for custom lint/breaking plugins since there's no plugins to migrate from <=v1.
 	// TODO: If we ever need v3, then we will have to deal with this.
-	client, err := bufcheck.NewClient(logger, bufcheck.NewRunnerProvider(wasm.UnimplementedRuntime))
+	client, err := bufcheck.NewClient(logger, bufcheck.NewLocalRunnerProvider(
+		wasm.UnimplementedRuntime,
+		bufplugin.NopPluginKeyProvider,
+		bufplugin.NopPluginDataProvider,
+	))
 	if err != nil {
 		return nil, err
 	}
diff --git a/private/buf/bufworkspace/workspace.go b/private/buf/bufworkspace/workspace.go
index 68ea77ac06..b27eb8441f 100644
--- a/private/buf/bufworkspace/workspace.go
+++ b/private/buf/bufworkspace/workspace.go
@@ -18,6 +18,7 @@ import (
 	"github.com/bufbuild/buf/private/bufpkg/bufconfig"
 	"github.com/bufbuild/buf/private/bufpkg/bufmodule"
 	"github.com/bufbuild/buf/private/bufpkg/bufparse"
+	"github.com/bufbuild/buf/private/bufpkg/bufplugin"
 	"github.com/bufbuild/buf/private/pkg/slicesext"
 )
 
@@ -72,8 +73,14 @@ type Workspace interface {
 	// detector ignoring these configs anyways.
 	GetBreakingConfigForOpaqueID(opaqueID string) bufconfig.BreakingConfig
 	// PluginConfigs gets the configured PluginConfigs of the Workspace.
+	//
+	// These come from buf.yaml files.
 	PluginConfigs() []bufconfig.PluginConfig
-	// ConfiguredDepModuleRefs returns the configured dependencies of the Workspace as ModuleRefs.
+	// RemotePluginKeys gets the remote PluginKeys of the Workspace.
+	//
+	// These come from buf.lock files.
+	RemotePluginKeys() []bufplugin.PluginKey
+	// ConfiguredDepModuleRefs returns the configured dependencies of the Workspace as Refs.
 	//
 	// These come from buf.yaml files.
 	//
@@ -105,6 +112,7 @@ type workspace struct {
 	opaqueIDToLintConfig     map[string]bufconfig.LintConfig
 	opaqueIDToBreakingConfig map[string]bufconfig.BreakingConfig
 	pluginConfigs            []bufconfig.PluginConfig
+	remotePluginKeys         []bufplugin.PluginKey
 	configuredDepModuleRefs  []bufparse.Ref
 
 	// If true, the workspace was created from v2 buf.yamls.
@@ -117,6 +125,7 @@ func newWorkspace(
 	opaqueIDToLintConfig map[string]bufconfig.LintConfig,
 	opaqueIDToBreakingConfig map[string]bufconfig.BreakingConfig,
 	pluginConfigs []bufconfig.PluginConfig,
+	remotePluginKeys []bufplugin.PluginKey,
 	configuredDepModuleRefs []bufparse.Ref,
 	isV2 bool,
 ) *workspace {
@@ -125,6 +134,7 @@ func newWorkspace(
 		opaqueIDToLintConfig:     opaqueIDToLintConfig,
 		opaqueIDToBreakingConfig: opaqueIDToBreakingConfig,
 		pluginConfigs:            pluginConfigs,
+		remotePluginKeys:         remotePluginKeys,
 		configuredDepModuleRefs:  configuredDepModuleRefs,
 		isV2:                     isV2,
 	}
@@ -142,6 +152,10 @@ func (w *workspace) PluginConfigs() []bufconfig.PluginConfig {
 	return slicesext.Copy(w.pluginConfigs)
 }
 
+func (w *workspace) RemotePluginKeys() []bufplugin.PluginKey {
+	return slicesext.Copy(w.remotePluginKeys)
+}
+
 func (w *workspace) ConfiguredDepModuleRefs() []bufparse.Ref {
 	return slicesext.Copy(w.configuredDepModuleRefs)
 }
diff --git a/private/buf/bufworkspace/workspace_provider.go b/private/buf/bufworkspace/workspace_provider.go
index a9811072cf..58152f94b3 100644
--- a/private/buf/bufworkspace/workspace_provider.go
+++ b/private/buf/bufworkspace/workspace_provider.go
@@ -25,6 +25,7 @@ import (
 	"github.com/bufbuild/buf/private/bufpkg/bufconfig"
 	"github.com/bufbuild/buf/private/bufpkg/bufmodule"
 	"github.com/bufbuild/buf/private/bufpkg/bufparse"
+	"github.com/bufbuild/buf/private/bufpkg/bufplugin"
 	"github.com/bufbuild/buf/private/pkg/normalpath"
 	"github.com/bufbuild/buf/private/pkg/slicesext"
 	"github.com/bufbuild/buf/private/pkg/slogext"
@@ -77,12 +78,14 @@ func NewWorkspaceProvider(
 	graphProvider bufmodule.GraphProvider,
 	moduleDataProvider bufmodule.ModuleDataProvider,
 	commitProvider bufmodule.CommitProvider,
+	pluginKeyProvider bufplugin.PluginKeyProvider,
 ) WorkspaceProvider {
 	return newWorkspaceProvider(
 		logger,
 		graphProvider,
 		moduleDataProvider,
 		commitProvider,
+		pluginKeyProvider,
 	)
 }
 
@@ -93,6 +96,10 @@ type workspaceProvider struct {
 	graphProvider      bufmodule.GraphProvider
 	moduleDataProvider bufmodule.ModuleDataProvider
 	commitProvider     bufmodule.CommitProvider
+
+	// pluginKeyProvider is only used for getting remote plugin keys for a single module
+	// when an override is specified.
+	pluginKeyProvider bufplugin.PluginKeyProvider
 }
 
 func newWorkspaceProvider(
@@ -100,12 +107,14 @@ func newWorkspaceProvider(
 	graphProvider bufmodule.GraphProvider,
 	moduleDataProvider bufmodule.ModuleDataProvider,
 	commitProvider bufmodule.CommitProvider,
+	pluginKeyProvider bufplugin.PluginKeyProvider,
 ) *workspaceProvider {
 	return &workspaceProvider{
 		logger:             logger,
 		graphProvider:      graphProvider,
 		moduleDataProvider: moduleDataProvider,
 		commitProvider:     commitProvider,
+		pluginKeyProvider:  pluginKeyProvider,
 	}
 }
 
@@ -136,8 +145,11 @@ func (w *workspaceProvider) GetWorkspaceForModuleKey(
 	targetModuleConfig := bufconfig.DefaultModuleConfigV1
 	// By default, there will be no plugin configs, however, similar to the lint and breaking
 	// configs, there may be an override, in which case, we need to populate the plugin configs
-	// from the override.
-	var pluginConfigs []bufconfig.PluginConfig
+	// from the override. Any remote plugin refs will be resolved by the pluginKeyProvider.
+	var (
+		pluginConfigs    []bufconfig.PluginConfig
+		remotePluginKeys []bufplugin.PluginKey
+	)
 	if config.configOverride != "" {
 		bufYAMLFile, err := bufconfig.GetBufYAMLFileForOverride(config.configOverride)
 		if err != nil {
@@ -150,7 +162,7 @@ func (w *workspaceProvider) GetWorkspaceForModuleKey(
 		case 1:
 			// If we have a single ModuleConfig, we assume that regardless of whether or not
 			// This ModuleConfig has a name, that this is what the user intends to associate
-			// with the tqrget module. This also handles the v1 case - v1 buf.yamls will always
+			// with the target module. This also handles the v1 case - v1 buf.yamls will always
 			// only have a single ModuleConfig, and it was expected pre-refactor that regardless
 			// of if the ModuleConfig had a name associated with it or not, the lint and breaking
 			// config that came from it would be associated.
@@ -172,6 +184,26 @@ func (w *workspaceProvider) GetWorkspaceForModuleKey(
 		}
 		if bufYAMLFile.FileVersion() == bufconfig.FileVersionV2 {
 			pluginConfigs = bufYAMLFile.PluginConfigs()
+			remotePluginRefs := slicesext.Filter(
+				slicesext.Map(pluginConfigs, func(pluginConfig bufconfig.PluginConfig) bufparse.Ref {
+					return pluginConfig.Ref()
+				}),
+				func(ref bufparse.Ref) bool {
+					return ref != nil
+				},
+			)
+			// Resolve the remote plugin keys for any remote plugins.
+			if len(remotePluginRefs) > 0 {
+				var err error
+				remotePluginKeys, err = w.pluginKeyProvider.GetPluginKeysForPluginRefs(
+					ctx,
+					remotePluginRefs,
+					bufplugin.DigestTypeP1,
+				)
+				if err != nil {
+					return nil, err
+				}
+			}
 		}
 	}
 
@@ -209,18 +241,18 @@ func (w *workspaceProvider) GetWorkspaceForModuleKey(
 		opaqueIDToLintConfig,
 		opaqueIDToBreakingConfig,
 		pluginConfigs,
+		remotePluginKeys,
 		nil,
 		false,
 	), nil
 }
 
-func (w *workspaceProvider) GetWorkspaceForBucket(
+func (w *workspaceProvider) getWorkspaceTargetingForBucket(
 	ctx context.Context,
 	bucket storage.ReadBucket,
 	bucketTargeting buftarget.BucketTargeting,
 	options ...WorkspaceBucketOption,
-) (Workspace, error) {
-	defer slogext.DebugProfile(w.logger)()
+) (*workspaceTargeting, error) {
 	config, err := newWorkspaceBucketConfig(options)
 	if err != nil {
 		return nil, err
@@ -232,7 +264,7 @@ func (w *workspaceProvider) GetWorkspaceForBucket(
 			return nil, err
 		}
 	}
-	workspaceTargeting, err := newWorkspaceTargeting(
+	return newWorkspaceTargeting(
 		ctx,
 		w.logger,
 		config,
@@ -241,6 +273,21 @@ func (w *workspaceProvider) GetWorkspaceForBucket(
 		overrideBufYAMLFile,
 		config.ignoreAndDisallowV1BufWorkYAMLs,
 	)
+}
+
+func (w *workspaceProvider) GetWorkspaceForBucket(
+	ctx context.Context,
+	bucket storage.ReadBucket,
+	bucketTargeting buftarget.BucketTargeting,
+	options ...WorkspaceBucketOption,
+) (Workspace, error) {
+	defer slogext.DebugProfile(w.logger)()
+	workspaceTargeting, err := w.getWorkspaceTargetingForBucket(
+		ctx,
+		bucket,
+		bucketTargeting,
+		options...,
+	)
 	if err != nil {
 		return nil, err
 	}
@@ -358,7 +405,8 @@ func (w *workspaceProvider) getWorkspaceForBucketAndModuleDirPathsV1Beta1OrV1(
 	return w.getWorkspaceForBucketModuleSet(
 		moduleSet,
 		v1WorkspaceTargeting.bucketIDToModuleConfig,
-		nil,
+		nil, // No plugin configs for v1
+		nil, // No remote plugin keys for v1
 		v1WorkspaceTargeting.allConfiguredDepModuleRefs,
 		false,
 	)
@@ -370,6 +418,7 @@ func (w *workspaceProvider) getWorkspaceForBucketBufYAMLV2(
 	v2Targeting *v2Targeting,
 ) (*workspace, error) {
 	moduleSetBuilder := bufmodule.NewModuleSetBuilder(ctx, w.logger, w.moduleDataProvider, w.commitProvider)
+	var remotePluginKeys []bufplugin.PluginKey
 	bufLockFile, err := bufconfig.GetBufLockFileForPrefix(
 		ctx,
 		bucket,
@@ -398,6 +447,7 @@ func (w *workspaceProvider) getWorkspaceForBucketBufYAMLV2(
 				false,
 			)
 		}
+		remotePluginKeys = bufLockFile.RemotePluginKeys()
 	}
 	// Only check for duplicate module description in v2, which would be an user error, i.e.
 	// This is not a system error:
@@ -455,6 +505,7 @@ func (w *workspaceProvider) getWorkspaceForBucketBufYAMLV2(
 		moduleSet,
 		v2Targeting.bucketIDToModuleConfig,
 		v2Targeting.bufYAMLFile.PluginConfigs(),
+		remotePluginKeys,
 		v2Targeting.bufYAMLFile.ConfiguredDepModuleRefs(),
 		true,
 	)
@@ -465,6 +516,7 @@ func (w *workspaceProvider) getWorkspaceForBucketModuleSet(
 	moduleSet bufmodule.ModuleSet,
 	bucketIDToModuleConfig map[string]bufconfig.ModuleConfig,
 	pluginConfigs []bufconfig.PluginConfig,
+	remotePluginKeys []bufplugin.PluginKey,
 	// Expected to already be unique by FullName.
 	configuredDepModuleRefs []bufparse.Ref,
 	isV2 bool,
@@ -490,6 +542,7 @@ func (w *workspaceProvider) getWorkspaceForBucketModuleSet(
 		opaqueIDToLintConfig,
 		opaqueIDToBreakingConfig,
 		pluginConfigs,
+		remotePluginKeys,
 		configuredDepModuleRefs,
 		isV2,
 	), nil
diff --git a/private/buf/bufworkspace/workspace_test.go b/private/buf/bufworkspace/workspace_test.go
index 0006c48217..01e59e5049 100644
--- a/private/buf/bufworkspace/workspace_test.go
+++ b/private/buf/bufworkspace/workspace_test.go
@@ -24,6 +24,7 @@ import (
 	"github.com/bufbuild/buf/private/buf/buftarget"
 	"github.com/bufbuild/buf/private/bufpkg/bufmodule"
 	"github.com/bufbuild/buf/private/bufpkg/bufmodule/bufmoduletesting"
+	"github.com/bufbuild/buf/private/bufpkg/bufplugin"
 	"github.com/bufbuild/buf/private/pkg/dag/dagtest"
 	"github.com/bufbuild/buf/private/pkg/ioext"
 	"github.com/bufbuild/buf/private/pkg/normalpath"
@@ -321,6 +322,7 @@ func testNewWorkspaceProvider(t *testing.T, testModuleDatas ...bufmoduletesting.
 		bsrProvider,
 		bsrProvider,
 		bsrProvider,
+		bufplugin.NopPluginKeyProvider,
 	)
 }
 
diff --git a/private/buf/cmd/buf/buf_test.go b/private/buf/cmd/buf/buf_test.go
index d5b0d13fcf..f5ec4472ce 100644
--- a/private/buf/cmd/buf/buf_test.go
+++ b/private/buf/cmd/buf/buf_test.go
@@ -35,6 +35,7 @@ import (
 	"github.com/bufbuild/buf/private/bufpkg/bufcheck"
 	"github.com/bufbuild/buf/private/bufpkg/bufconfig"
 	"github.com/bufbuild/buf/private/bufpkg/bufimage"
+	"github.com/bufbuild/buf/private/bufpkg/bufplugin"
 	imagev1 "github.com/bufbuild/buf/private/gen/proto/go/buf/alpha/image/v1"
 	"github.com/bufbuild/buf/private/pkg/app/appcmd"
 	"github.com/bufbuild/buf/private/pkg/app/appcmd/appcmdtesting"
@@ -1350,7 +1351,11 @@ func TestCheckLsBreakingRulesFromConfigExceptDeprecated(t *testing.T) {
 			// Do not need any custom lint/breaking plugins here.
 			client, err := bufcheck.NewClient(
 				slogtestext.NewLogger(t),
-				bufcheck.NewRunnerProvider(wasm.UnimplementedRuntime),
+				bufcheck.NewLocalRunnerProvider(
+					wasm.UnimplementedRuntime,
+					bufplugin.NopPluginKeyProvider,
+					bufplugin.NopPluginDataProvider,
+				),
 			)
 			require.NoError(t, err)
 			allRules, err := client.AllRules(context.Background(), check.RuleTypeBreaking, version)
diff --git a/private/buf/cmd/buf/command/beta/lsp/lsp.go b/private/buf/cmd/buf/command/beta/lsp/lsp.go
index 24f4bec061..e7b7b2fc4e 100644
--- a/private/buf/cmd/buf/command/beta/lsp/lsp.go
+++ b/private/buf/cmd/buf/command/beta/lsp/lsp.go
@@ -26,7 +26,6 @@ import (
 
 	"github.com/bufbuild/buf/private/buf/bufcli"
 	"github.com/bufbuild/buf/private/buf/buflsp"
-	"github.com/bufbuild/buf/private/bufpkg/bufcheck"
 	"github.com/bufbuild/buf/private/pkg/app/appcmd"
 	"github.com/bufbuild/buf/private/pkg/app/appext"
 	"github.com/bufbuild/buf/private/pkg/ioext"
@@ -113,16 +112,8 @@ func run(
 	defer func() {
 		retErr = errors.Join(retErr, wasmRuntime.Close(ctx))
 	}()
-	checkClient, err := bufcheck.NewClient(
-		container.Logger(),
-		bufcheck.NewRunnerProvider(wasmRuntime),
-		bufcheck.ClientWithStderr(container.Stderr()),
-	)
-	if err != nil {
-		return err
-	}
 
-	conn, err := buflsp.Serve(ctx, wktBucket, container, controller, checkClient, jsonrpc2.NewStream(transport))
+	conn, err := buflsp.Serve(ctx, wktBucket, container, controller, wasmRuntime, jsonrpc2.NewStream(transport))
 	if err != nil {
 		return err
 	}
diff --git a/private/buf/cmd/buf/command/breaking/breaking.go b/private/buf/cmd/buf/command/breaking/breaking.go
index 71b699c9a2..869cb3a84e 100644
--- a/private/buf/cmd/buf/command/breaking/breaking.go
+++ b/private/buf/cmd/buf/command/breaking/breaking.go
@@ -163,11 +163,14 @@ func run(
 	}
 	// Do not exclude imports here. bufcheck's Client requires all imports.
 	// Use bufcheck's BreakingWithExcludeImports.
+	inputControllerOptions := []bufctl.FunctionOption{
+		bufctl.WithTargetPaths(flags.Paths, flags.ExcludePaths),
+		bufctl.WithConfigOverride(flags.Config),
+	}
 	imageWithConfigs, err := controller.GetTargetImageWithConfigs(
 		ctx,
 		input,
-		bufctl.WithTargetPaths(flags.Paths, flags.ExcludePaths),
-		bufctl.WithConfigOverride(flags.Config),
+		inputControllerOptions...,
 	)
 	if err != nil {
 		return err
@@ -216,11 +219,20 @@ func run(
 	defer func() {
 		retErr = errors.Join(retErr, wasmRuntime.Close(ctx))
 	}()
+	checkRunnerProvider, err := controller.GetCheckRunnerProvider(
+		ctx,
+		input,
+		wasmRuntime,
+		inputControllerOptions...,
+	)
+	if err != nil {
+		return err
+	}
 	var allFileAnnotations []bufanalysis.FileAnnotation
 	for i, imageWithConfig := range imageWithConfigs {
 		client, err := bufcheck.NewClient(
 			container.Logger(),
-			bufcheck.NewRunnerProvider(wasmRuntime),
+			checkRunnerProvider,
 			bufcheck.ClientWithStderr(container.Stderr()),
 		)
 		if err != nil {
diff --git a/private/buf/cmd/buf/command/config/internal/internal.go b/private/buf/cmd/buf/command/config/internal/internal.go
index f1dfeb5524..40af99066c 100644
--- a/private/buf/cmd/buf/command/config/internal/internal.go
+++ b/private/buf/cmd/buf/command/config/internal/internal.go
@@ -183,6 +183,10 @@ func lsRun(
 			return err
 		}
 	}
+	controller, err := bufcli.NewController(container)
+	if err != nil {
+		return err
+	}
 	wasmRuntimeCacheDir, err := bufcli.CreateWasmRuntimeCacheDir(container)
 	if err != nil {
 		return err
@@ -194,9 +198,13 @@ func lsRun(
 	defer func() {
 		retErr = errors.Join(retErr, wasmRuntime.Close(ctx))
 	}()
+	checkRunnerProvider, err := controller.GetCheckRunnerProvider(ctx, ".", wasmRuntime)
+	if err != nil {
+		return err
+	}
 	client, err := bufcheck.NewClient(
 		container.Logger(),
-		bufcheck.NewRunnerProvider(wasmRuntime),
+		checkRunnerProvider,
 		bufcheck.ClientWithStderr(container.Stderr()),
 	)
 	if err != nil {
diff --git a/private/buf/cmd/buf/command/lint/lint.go b/private/buf/cmd/buf/command/lint/lint.go
index 496a0d6638..839558f346 100644
--- a/private/buf/cmd/buf/command/lint/lint.go
+++ b/private/buf/cmd/buf/command/lint/lint.go
@@ -121,11 +121,14 @@ func run(
 	if err != nil {
 		return err
 	}
+	controllerOptions := []bufctl.FunctionOption{
+		bufctl.WithTargetPaths(flags.Paths, flags.ExcludePaths),
+		bufctl.WithConfigOverride(flags.Config),
+	}
 	imageWithConfigs, err := controller.GetTargetImageWithConfigs(
 		ctx,
 		input,
-		bufctl.WithTargetPaths(flags.Paths, flags.ExcludePaths),
-		bufctl.WithConfigOverride(flags.Config),
+		controllerOptions...,
 	)
 	if err != nil {
 		return err
@@ -141,11 +144,20 @@ func run(
 	defer func() {
 		retErr = errors.Join(retErr, wasmRuntime.Close(ctx))
 	}()
+	checkRunnerProvider, err := controller.GetCheckRunnerProvider(
+		ctx,
+		input,
+		wasmRuntime,
+		controllerOptions...,
+	)
+	if err != nil {
+		return err
+	}
 	var allFileAnnotations []bufanalysis.FileAnnotation
 	for _, imageWithConfig := range imageWithConfigs {
 		client, err := bufcheck.NewClient(
 			container.Logger(),
-			bufcheck.NewRunnerProvider(wasmRuntime),
+			checkRunnerProvider,
 			bufcheck.ClientWithStderr(container.Stderr()),
 		)
 		if err != nil {
diff --git a/private/buf/cmd/buf/command/mod/internal/internal.go b/private/buf/cmd/buf/command/mod/internal/internal.go
index c802e34e50..5aa14892c4 100644
--- a/private/buf/cmd/buf/command/mod/internal/internal.go
+++ b/private/buf/cmd/buf/command/mod/internal/internal.go
@@ -24,6 +24,7 @@ import (
 	"github.com/bufbuild/buf/private/buf/bufcli"
 	"github.com/bufbuild/buf/private/bufpkg/bufcheck"
 	"github.com/bufbuild/buf/private/bufpkg/bufconfig"
+	"github.com/bufbuild/buf/private/bufpkg/bufplugin"
 	"github.com/bufbuild/buf/private/pkg/app/appcmd"
 	"github.com/bufbuild/buf/private/pkg/app/appext"
 	"github.com/bufbuild/buf/private/pkg/slicesext"
@@ -175,7 +176,11 @@ func lsRun(
 	// BufYAMLFiles <=v1 never had plugins.
 	client, err := bufcheck.NewClient(
 		container.Logger(),
-		bufcheck.NewRunnerProvider(wasm.UnimplementedRuntime),
+		bufcheck.NewLocalRunnerProvider(
+			wasm.UnimplementedRuntime,
+			bufplugin.NopPluginKeyProvider,
+			bufplugin.NopPluginDataProvider,
+		),
 		bufcheck.ClientWithStderr(container.Stderr()),
 	)
 	if err != nil {
diff --git a/private/buf/cmd/buf/command/plugin/pluginpush/pluginpush.go b/private/buf/cmd/buf/command/plugin/pluginpush/pluginpush.go
index 4f509d1c01..94efe36374 100644
--- a/private/buf/cmd/buf/command/plugin/pluginpush/pluginpush.go
+++ b/private/buf/cmd/buf/command/plugin/pluginpush/pluginpush.go
@@ -148,9 +148,13 @@ func upload(
 	var plugin bufplugin.Plugin
 	switch {
 	case flags.Binary != "":
+		// We create a local plugin reference to the Wasm binary.
+		pluginName := flags.Binary
 		var err error
 		plugin, err = bufplugin.NewLocalWasmPlugin(
 			pluginFullName,
+			pluginName,
+			nil, // args
 			func() ([]byte, error) {
 				wasmBinary, err := os.ReadFile(flags.Binary)
 				if err != nil {
diff --git a/private/buf/cmd/protoc-gen-buf-breaking/breaking.go b/private/buf/cmd/protoc-gen-buf-breaking/breaking.go
index 1de59b0149..82d8b856c2 100644
--- a/private/buf/cmd/protoc-gen-buf-breaking/breaking.go
+++ b/private/buf/cmd/protoc-gen-buf-breaking/breaking.go
@@ -28,6 +28,7 @@ import (
 	"github.com/bufbuild/buf/private/bufpkg/bufanalysis"
 	"github.com/bufbuild/buf/private/bufpkg/bufcheck"
 	"github.com/bufbuild/buf/private/bufpkg/bufimage"
+	"github.com/bufbuild/buf/private/bufpkg/bufplugin"
 	"github.com/bufbuild/buf/private/pkg/encoding"
 	"github.com/bufbuild/buf/private/pkg/protodescriptor"
 	"github.com/bufbuild/buf/private/pkg/protoencoding"
@@ -125,7 +126,11 @@ func handle(
 	// The protoc plugins do not support custom lint/breaking change plugins for now.
 	client, err := bufcheck.NewClient(
 		container.Logger(),
-		bufcheck.NewRunnerProvider(wasm.UnimplementedRuntime),
+		bufcheck.NewLocalRunnerProvider(
+			wasm.UnimplementedRuntime,
+			bufplugin.NopPluginKeyProvider,
+			bufplugin.NopPluginDataProvider,
+		),
 		bufcheck.ClientWithStderr(pluginEnv.Stderr),
 	)
 	if err != nil {
diff --git a/private/buf/cmd/protoc-gen-buf-lint/lint.go b/private/buf/cmd/protoc-gen-buf-lint/lint.go
index d7ed7b0a1a..f29fef7659 100644
--- a/private/buf/cmd/protoc-gen-buf-lint/lint.go
+++ b/private/buf/cmd/protoc-gen-buf-lint/lint.go
@@ -27,6 +27,7 @@ import (
 	"github.com/bufbuild/buf/private/bufpkg/bufanalysis"
 	"github.com/bufbuild/buf/private/bufpkg/bufcheck"
 	"github.com/bufbuild/buf/private/bufpkg/bufimage"
+	"github.com/bufbuild/buf/private/bufpkg/bufplugin"
 	"github.com/bufbuild/buf/private/pkg/encoding"
 	"github.com/bufbuild/buf/private/pkg/protodescriptor"
 	"github.com/bufbuild/buf/private/pkg/protoencoding"
@@ -100,7 +101,11 @@ func handle(
 	// The protoc plugins do not support custom lint/breaking change plugins for now.
 	client, err := bufcheck.NewClient(
 		container.Logger(),
-		bufcheck.NewRunnerProvider(wasm.UnimplementedRuntime),
+		bufcheck.NewLocalRunnerProvider(
+			wasm.UnimplementedRuntime,
+			bufplugin.NopPluginKeyProvider,
+			bufplugin.NopPluginDataProvider,
+		),
 		bufcheck.ClientWithStderr(pluginEnv.Stderr),
 	)
 	if err != nil {
diff --git a/private/bufpkg/bufcheck/breaking_test.go b/private/bufpkg/bufcheck/breaking_test.go
index 2bab7cbf4d..32b4716baa 100644
--- a/private/bufpkg/bufcheck/breaking_test.go
+++ b/private/bufpkg/bufcheck/breaking_test.go
@@ -30,6 +30,7 @@ import (
 	"github.com/bufbuild/buf/private/bufpkg/bufcheck"
 	"github.com/bufbuild/buf/private/bufpkg/bufimage"
 	"github.com/bufbuild/buf/private/bufpkg/bufmodule"
+	"github.com/bufbuild/buf/private/bufpkg/bufplugin"
 	"github.com/bufbuild/buf/private/pkg/slogtestext"
 	"github.com/bufbuild/buf/private/pkg/storage/storageos"
 	"github.com/bufbuild/buf/private/pkg/wasm"
@@ -1312,6 +1313,7 @@ func testBreaking(
 		bufmodule.NopGraphProvider,
 		bufmodule.NopModuleDataProvider,
 		bufmodule.NopCommitProvider,
+		bufplugin.NopPluginKeyProvider,
 	)
 	previousWorkspace, err := workspaceProvider.GetWorkspaceForBucket(
 		ctx,
@@ -1344,7 +1346,11 @@ func testBreaking(
 	require.NoError(t, err)
 	breakingConfig := workspace.GetBreakingConfigForOpaqueID(opaqueID)
 	require.NotNil(t, breakingConfig)
-	client, err := bufcheck.NewClient(logger, bufcheck.NewRunnerProvider(wasm.UnimplementedRuntime))
+	client, err := bufcheck.NewClient(logger, bufcheck.NewLocalRunnerProvider(
+		wasm.UnimplementedRuntime,
+		bufplugin.NopPluginKeyProvider,
+		bufplugin.NopPluginDataProvider,
+	))
 	require.NoError(t, err)
 	err = client.Breaking(
 		ctx,
diff --git a/private/bufpkg/bufcheck/bufcheck.go b/private/bufpkg/bufcheck/bufcheck.go
index 6036e7b401..d5cee8cfc9 100644
--- a/private/bufpkg/bufcheck/bufcheck.go
+++ b/private/bufpkg/bufcheck/bufcheck.go
@@ -22,6 +22,7 @@ import (
 	"buf.build/go/bufplugin/check"
 	"github.com/bufbuild/buf/private/bufpkg/bufconfig"
 	"github.com/bufbuild/buf/private/bufpkg/bufimage"
+	"github.com/bufbuild/buf/private/bufpkg/bufplugin"
 	"github.com/bufbuild/buf/private/pkg/slicesext"
 	"github.com/bufbuild/buf/private/pkg/syserror"
 	"github.com/bufbuild/buf/private/pkg/wasm"
@@ -169,7 +170,8 @@ func (r RunnerProviderFunc) NewRunner(pluginConfig bufconfig.PluginConfig) (plug
 	return r(pluginConfig)
 }
 
-// NewRunnerProvider returns a new RunnerProvider for the wasm.Runtime.
+// NewLocalRunnerProvider returns a new RunnerProvider for the wasm.Runtime and
+// the given plugin providers.
 //
 // This implementation should only be used for local applications. It is safe to
 // use concurrently.
@@ -178,13 +180,22 @@ func (r RunnerProviderFunc) NewRunner(pluginConfig bufconfig.PluginConfig) (plug
 // The supported types are:
 //   - bufconfig.PluginConfigTypeLocal
 //   - bufconfig.PluginConfigTypeLocalWasm
+//   - bufconfig.PluginConfigTypeRemoteWasm
 //
 // If the PluginConfigType is not supported, an error is returned.
-func NewRunnerProvider(
+// To disable support for Wasm plugins, set wasmRuntime to wasm.UnimplementedRuntime.
+// To disable support for bufconfig.PluginConfigTypeRemoteWasm Plugins, set
+// pluginKeyProvider and pluginDataProvider to bufplugin.NopPluginKeyProvider
+// and bufplugin.NopPluginDataProvider.
+func NewLocalRunnerProvider(
 	wasmRuntime wasm.Runtime,
+	pluginKeyProvider bufplugin.PluginKeyProvider,
+	pluginDataProvider bufplugin.PluginDataProvider,
 ) RunnerProvider {
 	return newRunnerProvider(
 		wasmRuntime,
+		pluginKeyProvider,
+		pluginDataProvider,
 	)
 }
 
diff --git a/private/bufpkg/bufcheck/lint_test.go b/private/bufpkg/bufcheck/lint_test.go
index 626f3738b2..cbc8ce64cd 100644
--- a/private/bufpkg/bufcheck/lint_test.go
+++ b/private/bufpkg/bufcheck/lint_test.go
@@ -27,6 +27,7 @@ import (
 	"github.com/bufbuild/buf/private/bufpkg/bufcheck"
 	"github.com/bufbuild/buf/private/bufpkg/bufimage"
 	"github.com/bufbuild/buf/private/bufpkg/bufmodule"
+	"github.com/bufbuild/buf/private/bufpkg/bufplugin"
 	"github.com/bufbuild/buf/private/pkg/slogtestext"
 	"github.com/bufbuild/buf/private/pkg/storage/storageos"
 	"github.com/bufbuild/buf/private/pkg/wasm"
@@ -1318,6 +1319,7 @@ func testLintWithOptions(
 		bufmodule.NopGraphProvider,
 		bufmodule.NopModuleDataProvider,
 		bufmodule.NopCommitProvider,
+		bufplugin.NopPluginKeyProvider,
 	).GetWorkspaceForBucket(
 		ctx,
 		readWriteBucket,
@@ -1355,7 +1357,11 @@ func testLintWithOptions(
 	})
 	client, err := bufcheck.NewClient(
 		logger,
-		bufcheck.NewRunnerProvider(wasmRuntime),
+		bufcheck.NewLocalRunnerProvider(
+			wasmRuntime,
+			bufplugin.NopPluginKeyProvider,
+			bufplugin.NopPluginDataProvider,
+		),
 	)
 	require.NoError(t, err)
 	err = client.Lint(
diff --git a/private/bufpkg/bufcheck/multi_client_test.go b/private/bufpkg/bufcheck/multi_client_test.go
index 8fde74ddae..895ffe85dd 100644
--- a/private/bufpkg/bufcheck/multi_client_test.go
+++ b/private/bufpkg/bufcheck/multi_client_test.go
@@ -24,6 +24,7 @@ import (
 	"buf.build/go/bufplugin/check/checkutil"
 	"buf.build/go/bufplugin/option"
 	"github.com/bufbuild/buf/private/bufpkg/bufconfig"
+	"github.com/bufbuild/buf/private/bufpkg/bufplugin"
 	"github.com/bufbuild/buf/private/pkg/slicesext"
 	"github.com/bufbuild/buf/private/pkg/slogtestext"
 	"github.com/bufbuild/buf/private/pkg/stringutil"
@@ -182,13 +183,17 @@ func TestMultiClientCannotHaveOverlappingRulesWithBuiltIn(t *testing.T) {
 
 	client, err := newClient(
 		slogtestext.NewLogger(t),
-		NewRunnerProvider(wasm.UnimplementedRuntime),
+		NewLocalRunnerProvider(
+			wasm.UnimplementedRuntime,
+			bufplugin.NopPluginKeyProvider,
+			bufplugin.NopPluginDataProvider,
+		),
 	)
 	require.NoError(t, err)
 	duplicateBuiltInRulePluginConfig, err := bufconfig.NewLocalPluginConfig(
 		"buf-plugin-duplicate-rule",
 		nil,
-		[]string{"buf-plugin-duplicate-rule"},
+		nil,
 	)
 	require.NoError(t, err)
 	emptyOptions, err := option.NewOptions(nil)
@@ -275,13 +280,17 @@ func TestMultiClientCannotHaveOverlappingCategoriesWithBuiltIn(t *testing.T) {
 
 	client, err := newClient(
 		slogtestext.NewLogger(t),
-		NewRunnerProvider(wasm.UnimplementedRuntime),
+		NewLocalRunnerProvider(
+			wasm.UnimplementedRuntime,
+			bufplugin.NopPluginKeyProvider,
+			bufplugin.NopPluginDataProvider,
+		),
 	)
 	require.NoError(t, err)
 	duplicateBuiltInRulePluginConfig, err := bufconfig.NewLocalPluginConfig(
 		"buf-plugin-duplicate-category",
 		nil,
-		[]string{"buf-plugin-duplicate-category"},
+		nil,
 	)
 	require.NoError(t, err)
 	emptyOptions, err := option.NewOptions(nil)
diff --git a/private/bufpkg/bufcheck/runner_provider.go b/private/bufpkg/bufcheck/runner_provider.go
index 013c1a892f..3e7d73c3dd 100644
--- a/private/bufpkg/bufcheck/runner_provider.go
+++ b/private/bufpkg/bufcheck/runner_provider.go
@@ -15,9 +15,12 @@
 package bufcheck
 
 import (
-	"fmt"
+	"context"
+	"sync"
 
 	"github.com/bufbuild/buf/private/bufpkg/bufconfig"
+	"github.com/bufbuild/buf/private/bufpkg/bufparse"
+	"github.com/bufbuild/buf/private/bufpkg/bufplugin"
 	"github.com/bufbuild/buf/private/pkg/pluginrpcutil"
 	"github.com/bufbuild/buf/private/pkg/syserror"
 	"github.com/bufbuild/buf/private/pkg/wasm"
@@ -25,37 +28,135 @@ import (
 )
 
 type runnerProvider struct {
-	wasmRuntime wasm.Runtime
+	wasmRuntime        wasm.Runtime
+	pluginKeyProvider  bufplugin.PluginKeyProvider
+	pluginDataProvider bufplugin.PluginDataProvider
 }
 
 func newRunnerProvider(
 	wasmRuntime wasm.Runtime,
+	pluginKeyProvider bufplugin.PluginKeyProvider,
+	pluginDataProvider bufplugin.PluginDataProvider,
 ) *runnerProvider {
 	return &runnerProvider{
-		wasmRuntime: wasmRuntime,
+		wasmRuntime:        wasmRuntime,
+		pluginKeyProvider:  pluginKeyProvider,
+		pluginDataProvider: pluginDataProvider,
 	}
 }
 
 func (r *runnerProvider) NewRunner(pluginConfig bufconfig.PluginConfig) (pluginrpc.Runner, error) {
 	switch pluginConfig.Type() {
 	case bufconfig.PluginConfigTypeLocal:
-		path := pluginConfig.Path()
-		return pluginrpcutil.NewRunner(
-			// We know that Path is of at least length 1.
-			path[0],
-			path[1:]...,
+		return pluginrpcutil.NewLocalRunner(
+			pluginConfig.Name(),
+			pluginConfig.Args()...,
 		), nil
 	case bufconfig.PluginConfigTypeLocalWasm:
-		path := pluginConfig.Path()
-		return pluginrpcutil.NewWasmRunner(
+		return pluginrpcutil.NewLocalWasmRunner(
 			r.wasmRuntime,
-			// We know that Path is of at least length 1.
-			path[0],
-			path[1:]...,
+			pluginConfig.Name(),
+			pluginConfig.Args()...,
 		), nil
-	case bufconfig.PluginConfigTypeRemote:
-		return nil, fmt.Errorf("remote plugins are not supported")
+	case bufconfig.PluginConfigTypeRemoteWasm:
+		return newRemoteWasmPluginRunner(
+			r.wasmRuntime,
+			r.pluginKeyProvider,
+			r.pluginDataProvider,
+			pluginConfig,
+		)
 	default:
 		return nil, syserror.Newf("unknown PluginConfigType: %v", pluginConfig.Type())
 	}
 }
+
+// *** PRIVATE ***
+
+// remoteWasmPluginRunner is a Runner that runs a remote Wasm plugin.
+//
+// This is a wrapper around a pluginrpc.Runner that first resolves the Ref to
+// a PluginKey using the PluginKeyProvider. It then loads the PluginData for
+// the PluginKey using the PluginDataProvider. The PluginData is then used to
+// create the pluginrpc.Runner. The Runner is only loaded once and is cached
+// for future calls. However, if the Runner fails to load it will try to
+// reload on the next call.
+type remoteWasmPluginRunner struct {
+	wasmRuntime        wasm.Runtime
+	pluginKeyProvider  bufplugin.PluginKeyProvider
+	pluginDataProvider bufplugin.PluginDataProvider
+	pluginRef          bufparse.Ref
+	pluginName         string
+	pluginArgs         []string
+	// lock protects runner.
+	lock   sync.RWMutex
+	runner pluginrpc.Runner
+}
+
+func newRemoteWasmPluginRunner(
+	wasmRuntime wasm.Runtime,
+	pluginKeyProvider bufplugin.PluginKeyProvider,
+	pluginDataProvider bufplugin.PluginDataProvider,
+	pluginConfig bufconfig.PluginConfig,
+) (*remoteWasmPluginRunner, error) {
+	pluginRef := pluginConfig.Ref()
+	if pluginRef == nil {
+		return nil, syserror.Newf("Ref nil on PluginConfig of type %v", bufconfig.PluginConfigTypeRemoteWasm)
+	}
+	return &remoteWasmPluginRunner{
+		wasmRuntime:        wasmRuntime,
+		pluginKeyProvider:  pluginKeyProvider,
+		pluginDataProvider: pluginDataProvider,
+		pluginRef:          pluginRef,
+		pluginName:         pluginConfig.Name(),
+		pluginArgs:         pluginConfig.Args(),
+	}, nil
+}
+
+func (r *remoteWasmPluginRunner) Run(ctx context.Context, env pluginrpc.Env) (retErr error) {
+	delegate, err := r.loadRunnerOnce(ctx)
+	if err != nil {
+		return err
+	}
+	return delegate.Run(ctx, env)
+}
+
+func (r *remoteWasmPluginRunner) loadRunnerOnce(ctx context.Context) (pluginrpc.Runner, error) {
+	r.lock.RLock()
+	if r.runner != nil {
+		r.lock.RUnlock()
+		return r.runner, nil
+	}
+	r.lock.RUnlock()
+	r.lock.Lock()
+	defer r.lock.Unlock()
+	if r.runner == nil {
+		runner, err := r.loadRunner(ctx)
+		if err != nil {
+			// The error isn't stored to avoid ctx cancellation issues. On the next call,
+			// the runner will be reloaded instead of returning the erorr.
+			return nil, err
+		}
+		r.runner = runner
+	}
+	return r.runner, nil
+}
+
+func (r *remoteWasmPluginRunner) loadRunner(ctx context.Context) (pluginrpc.Runner, error) {
+	pluginKeys, err := r.pluginKeyProvider.GetPluginKeysForPluginRefs(ctx, []bufparse.Ref{r.pluginRef}, bufplugin.DigestTypeP1)
+	if err != nil {
+		return nil, err
+	}
+	if len(pluginKeys) != 1 {
+		return nil, syserror.Newf("expected 1 PluginKey, got %d", len(pluginKeys))
+	}
+	// Load the data for the plugin now to ensure the context is valid for the entire operation.
+	pluginDatas, err := r.pluginDataProvider.GetPluginDatasForPluginKeys(ctx, pluginKeys)
+	if err != nil {
+		return nil, err
+	}
+	if len(pluginDatas) != 1 {
+		return nil, syserror.Newf("expected 1 PluginData, got %d", len(pluginDatas))
+	}
+	data := pluginDatas[0]
+	return pluginrpcutil.NewWasmRunner(r.wasmRuntime, data.Data, r.pluginName, r.pluginArgs...), nil
+}
diff --git a/private/bufpkg/bufconfig/buf_lock_file.go b/private/bufpkg/bufconfig/buf_lock_file.go
index bb0490fd1e..c37abd22a9 100644
--- a/private/bufpkg/bufconfig/buf_lock_file.go
+++ b/private/bufpkg/bufconfig/buf_lock_file.go
@@ -205,13 +205,22 @@ func newBufLockFile(
 	if err := validateNoDuplicateModuleKeysByFullName(depModuleKeys); err != nil {
 		return nil, err
 	}
+	if err := validateNoDuplicatePluginKeysByFullName(remotePluginKeys); err != nil {
+		return nil, err
+	}
 	switch fileVersion {
 	case FileVersionV1Beta1, FileVersionV1:
-		if err := validateExpectedDigestType(depModuleKeys, fileVersion, bufmodule.DigestTypeB4); err != nil {
+		if err := validateModuleExpectedDigestType(depModuleKeys, fileVersion, bufmodule.DigestTypeB4); err != nil {
 			return nil, err
 		}
+		if len(remotePluginKeys) > 0 {
+			return nil, errors.New("remote plugins are not supported in v1 or v1beta1 buf.lock files")
+		}
 	case FileVersionV2:
-		if err := validateExpectedDigestType(depModuleKeys, fileVersion, bufmodule.DigestTypeB5); err != nil {
+		if err := validateModuleExpectedDigestType(depModuleKeys, fileVersion, bufmodule.DigestTypeB5); err != nil {
+			return nil, err
+		}
+		if err := validatePluginExpectedDigestType(remotePluginKeys, fileVersion, bufplugin.DigestTypeP1); err != nil {
 			return nil, err
 		}
 	default:
@@ -522,6 +531,18 @@ func validateNoDuplicateModuleKeysByFullName(moduleKeys []bufmodule.ModuleKey) e
 	return nil
 }
 
+func validateNoDuplicatePluginKeysByFullName(pluginKeys []bufplugin.PluginKey) error {
+	pluginFullNameStringMap := make(map[string]struct{})
+	for _, pluginKey := range pluginKeys {
+		pluginFullNameString := pluginKey.FullName().String()
+		if _, ok := pluginFullNameStringMap[pluginFullNameString]; ok {
+			return fmt.Errorf("duplicate plugin %q attempted to be added to lock file", pluginFullNameString)
+		}
+		pluginFullNameStringMap[pluginFullNameString] = struct{}{}
+	}
+	return nil
+}
+
 func validateV1AndV1Beta1DepsHaveCommits(bufLockFile BufLockFile) error {
 	switch fileVersion := bufLockFile.FileVersion(); fileVersion {
 	case FileVersionV1Beta1, FileVersionV1:
@@ -545,7 +566,7 @@ func validateV1AndV1Beta1DepsHaveCommits(bufLockFile BufLockFile) error {
 	}
 }
 
-func validateExpectedDigestType(
+func validateModuleExpectedDigestType(
 	moduleKeys []bufmodule.ModuleKey,
 	fileVersion FileVersion,
 	expectedDigestType bufmodule.DigestType,
@@ -568,6 +589,29 @@ func validateExpectedDigestType(
 	return nil
 }
 
+func validatePluginExpectedDigestType(
+	pluginKeys []bufplugin.PluginKey,
+	fileVersion FileVersion,
+	expectedDigestType bufplugin.DigestType,
+) error {
+	for _, pluginKey := range pluginKeys {
+		digest, err := pluginKey.Digest()
+		if err != nil {
+			return err
+		}
+		if digest.Type() != expectedDigestType {
+			return fmt.Errorf(
+				"%s lock files must use digest type %v, but remote plugin %s had a digest type of %v",
+				fileVersion,
+				expectedDigestType,
+				pluginKey.String(),
+				digest.Type(),
+			)
+		}
+	}
+	return nil
+}
+
 // externalBufLockFileV1Beta1V1 represents the v1 or v1beta1 buf.lock file,
 // which have the same shape.
 type externalBufLockFileV1Beta1V1 struct {
diff --git a/private/bufpkg/bufconfig/plugin_config.go b/private/bufpkg/bufconfig/plugin_config.go
index d35f9f55da..967e4f611b 100644
--- a/private/bufpkg/bufconfig/plugin_config.go
+++ b/private/bufpkg/bufconfig/plugin_config.go
@@ -19,7 +19,6 @@ import (
 	"fmt"
 	"os"
 	"path/filepath"
-	"strings"
 
 	"github.com/bufbuild/buf/private/bufpkg/bufparse"
 	"github.com/bufbuild/buf/private/pkg/encoding"
@@ -31,8 +30,8 @@ const (
 	PluginConfigTypeLocal PluginConfigType = iota + 1
 	// PluginConfigTypeLocalWasm is the local Wasm plugin config type.
 	PluginConfigTypeLocalWasm
-	// PluginConfigTypeRemote is the remote plugin config type.
-	PluginConfigTypeRemote
+	// PluginConfigTypeRemoteWasm is the remote Wasm plugin config type.
+	PluginConfigTypeRemoteWasm
 )
 
 // PluginConfigType is a generate plugin configuration type.
@@ -42,17 +41,22 @@ type PluginConfigType int
 type PluginConfig interface {
 	// Type returns the plugin type. This is never the zero value.
 	Type() PluginConfigType
-	// Name returns the plugin name. This is never empty.
+	// Name returns the plugin name.
+	//
+	// This may be a path, a remote reference, or a Wasm file. Invoking code
+	// should check the Type to determine how to interpret this.
+	//
+	// This is never empty.
 	Name() string
 	// Options returns the plugin options.
 	//
 	// TODO: Will want good validation and good error messages for what this decodes.
 	// Otherwise we will confuse users. Do QA.
 	Options() map[string]any
-	// Path returns the path, including arguments, to invoke the binary plugin.
+	// Args returns the arguments, to invoke the plugin.
 	//
-	// This is not empty only when the plugin is local.
-	Path() []string
+	// This may be empty.
+	Args() []string
 	// Ref returns the plugin reference.
 	//
 	// This is only non-nil when the plugin is remote.
@@ -64,30 +68,47 @@ type PluginConfig interface {
 // NewLocalPluginConfig returns a new PluginConfig for a local plugin.
 func NewLocalPluginConfig(
 	name string,
+	args []string,
 	options map[string]any,
-	path []string,
 ) (PluginConfig, error) {
 	return newLocalPluginConfig(
 		name,
+		args,
 		options,
-		path,
 	)
 }
 
 // NewLocalWasmPluginConfig returns a new PluginConfig for a local Wasm plugin.
 //
-// The first path argument is the path to the Wasm plugin and must end with .wasm.
-// The remaining path arguments are the arguments to the Wasm plugin. These are passed
-// to the Wasm plugin as command line arguments.
+// The name is the path to the Wasm plugin and must end with .wasm.
+// The args are the arguments to the Wasm plugin. These are passed to the Wasm plugin
+// as command line arguments.
 func NewLocalWasmPluginConfig(
 	name string,
+	args []string,
 	options map[string]any,
-	path []string,
 ) (PluginConfig, error) {
 	return newLocalWasmPluginConfig(
 		name,
+		args,
+		options,
+	)
+}
+
+// NewRemoteWasmPluginConfig returns a new PluginConfig for a remote Wasm plugin.
+//
+// The pluginRef is the remote reference to the plugin.
+// The args are the arguments to the remote plugin. These are passed to the remote plugin
+// as command line arguments.
+func NewRemoteWasmPluginConfig(
+	pluginRef bufparse.Ref,
+	args []string,
+	options map[string]any,
+) (PluginConfig, error) {
+	return newRemotePluginConfig(
+		pluginRef,
+		args,
 		options,
-		path,
 	)
 }
 
@@ -95,9 +116,9 @@ func NewLocalWasmPluginConfig(
 
 type pluginConfig struct {
 	pluginConfigType PluginConfigType
-	name             string
 	options          map[string]any
-	path             []string
+	name             string
+	args             []string
 	ref              bufparse.Ref
 }
 
@@ -123,6 +144,7 @@ func newPluginConfigForExternalV2(
 	if len(path) == 0 {
 		return nil, errors.New("must specify a path to the plugin")
 	}
+	name, args := path[0], path[1:]
 	// Remote plugins are specified as plugin references.
 	if pluginRef, err := bufparse.ParseRef(path[0]); err == nil {
 		// Check if the local filepath exists, if it does presume its
@@ -131,6 +153,7 @@ func newPluginConfigForExternalV2(
 		if _, err := os.Stat(path[0]); os.IsNotExist(err) {
 			return newRemotePluginConfig(
 				pluginRef,
+				args,
 				options,
 			)
 		}
@@ -138,61 +161,63 @@ func newPluginConfigForExternalV2(
 	// Wasm plugins are suffixed with .wasm. Otherwise, it's a binary.
 	if filepath.Ext(path[0]) == ".wasm" {
 		return newLocalWasmPluginConfig(
-			strings.Join(path, " "),
+			name,
+			args,
 			options,
-			path,
 		)
 	}
 	return newLocalPluginConfig(
-		strings.Join(path, " "),
+		name,
+		args,
 		options,
-		path,
 	)
 }
 
 func newLocalPluginConfig(
 	name string,
+	args []string,
 	options map[string]any,
-	path []string,
 ) (*pluginConfig, error) {
-	if len(path) == 0 {
+	if len(name) == 0 {
 		return nil, errors.New("must specify a path to the plugin")
 	}
 	return &pluginConfig{
 		pluginConfigType: PluginConfigTypeLocal,
-		name:             name,
 		options:          options,
-		path:             path,
+		name:             name,
+		args:             args,
 	}, nil
 }
 
 func newLocalWasmPluginConfig(
 	name string,
+	args []string,
 	options map[string]any,
-	path []string,
 ) (*pluginConfig, error) {
-	if len(path) == 0 {
+	if len(name) == 0 {
 		return nil, errors.New("must specify a path to the plugin")
 	}
-	if filepath.Ext(path[0]) != ".wasm" {
+	if filepath.Ext(name) != ".wasm" {
 		return nil, fmt.Errorf("must specify a path to the plugin, and the first path argument must end with .wasm")
 	}
 	return &pluginConfig{
 		pluginConfigType: PluginConfigTypeLocalWasm,
-		name:             name,
 		options:          options,
-		path:             path,
+		name:             name,
+		args:             args,
 	}, nil
 }
 
 func newRemotePluginConfig(
 	pluginRef bufparse.Ref,
+	args []string,
 	options map[string]any,
 ) (*pluginConfig, error) {
 	return &pluginConfig{
-		pluginConfigType: PluginConfigTypeRemote,
-		name:             pluginRef.FullName().Name(),
+		pluginConfigType: PluginConfigTypeRemoteWasm,
+		name:             pluginRef.String(),
 		options:          options,
+		args:             args,
 		ref:              pluginRef,
 	}, nil
 }
@@ -209,8 +234,8 @@ func (p *pluginConfig) Options() map[string]any {
 	return p.options
 }
 
-func (p *pluginConfig) Path() []string {
-	return p.path
+func (p *pluginConfig) Args() []string {
+	return p.args
 }
 
 func (p *pluginConfig) Ref() bufparse.Ref {
@@ -229,15 +254,12 @@ func newExternalV2ForPluginConfig(
 	externalBufYAMLFilePluginV2 := externalBufYAMLFilePluginV2{
 		Options: pluginConfig.Options(),
 	}
-	switch pluginConfig.Type() {
-	case PluginConfigTypeLocal:
-		path := pluginConfig.Path()
-		switch {
-		case len(path) == 1:
-			externalBufYAMLFilePluginV2.Plugin = path[0]
-		case len(path) > 1:
-			externalBufYAMLFilePluginV2.Plugin = path
-		}
+	args := pluginConfig.Args()
+	switch {
+	case len(args) == 0:
+		externalBufYAMLFilePluginV2.Plugin = pluginConfig.Name()
+	case len(args) > 0:
+		externalBufYAMLFilePluginV2.Plugin = append([]string{pluginConfig.Name()}, args...)
 	}
 	return externalBufYAMLFilePluginV2, nil
 }
diff --git a/private/bufpkg/bufplugin/plugin.go b/private/bufpkg/bufplugin/plugin.go
index df03de8060..570e8fc2b8 100644
--- a/private/bufpkg/bufplugin/plugin.go
+++ b/private/bufpkg/bufplugin/plugin.go
@@ -37,10 +37,14 @@ type Plugin interface {
 	//
 	// If two Plugins have the same FullName, they will have the same OpaqueID.
 	OpaqueID() string
-	// Path returns the path, including arguments, to invoke the binary plugin.
+	// Name returns the name of the Plugin.
 	//
-	// This is not empty only when the Plugin is local.
-	Path() []string
+	// This is never empty.
+	Name() string
+	// Args returns the arguments to invoke the Plugin.
+	//
+	// This may be empty.
+	Args() []string
 	// FullName returns the full name of the Plugin.
 	//
 	// May be nil. Callers should not rely on this value being present.
@@ -104,15 +108,35 @@ type Plugin interface {
 	isPlugin()
 }
 
+// NewLocalPlugin returns a new Plugin for a local plugin.
+func NewLocalPlugin(
+	name string,
+	args []string,
+) (Plugin, error) {
+	return newPlugin(
+		"",  // description
+		nil, // pluginFullName
+		name,
+		args,
+		uuid.Nil, // commitID
+		false,    // isWasm
+		true,     // isLocal
+		nil,      // getData
+	)
+}
+
 // NewLocalWasmPlugin returns a new Plugin for a local Wasm plugin.
 func NewLocalWasmPlugin(
-	pluginFullName bufparse.FullName,
+	fullName bufparse.FullName,
+	name string,
+	args []string,
 	getData func() ([]byte, error),
 ) (Plugin, error) {
 	return newPlugin(
 		"", // description
-		pluginFullName,
-		nil,      // path
+		fullName,
+		name,
+		args,
 		uuid.Nil, // commitID
 		true,     // isWasm
 		true,     // isLocal
@@ -120,12 +144,32 @@ func NewLocalWasmPlugin(
 	)
 }
 
+// NewRemoteWasmPlugin returns a new Plugin for a remote Wasm plugin.
+func NewRemoteWasmPlugin(
+	pluginFullName bufparse.FullName,
+	args []string,
+	commitID uuid.UUID,
+	getData func() ([]byte, error),
+) (Plugin, error) {
+	return newPlugin(
+		"", // description
+		pluginFullName,
+		pluginFullName.String(),
+		args,
+		commitID,
+		true,  // isWasm
+		false, // isLocal
+		getData,
+	)
+}
+
 // *** PRIVATE ***
 
 type plugin struct {
 	description    string
 	pluginFullName bufparse.FullName
-	path           []string
+	name           string
+	args           []string
 	commitID       uuid.UUID
 	isWasm         bool
 	isLocal        bool
@@ -137,18 +181,19 @@ type plugin struct {
 func newPlugin(
 	description string,
 	pluginFullName bufparse.FullName,
-	path []string,
+	name string,
+	args []string,
 	commitID uuid.UUID,
 	isWasm bool,
 	isLocal bool,
 	getData func() ([]byte, error),
 ) (*plugin, error) {
+	if name == "" {
+		return nil, syserror.New("name not present when constructing a Plugin")
+	}
 	if isWasm && getData == nil {
 		return nil, syserror.Newf("getData not present when constructing a Wasm Plugin")
 	}
-	if !isWasm && len(path) == 0 {
-		return nil, syserror.New("path not present when constructing a non-Wasm Plugin")
-	}
 	if !isLocal && pluginFullName == nil {
 		return nil, syserror.New("pluginFullName not present when constructing a remote Plugin")
 	}
@@ -164,7 +209,8 @@ func newPlugin(
 	plugin := &plugin{
 		description:    description,
 		pluginFullName: pluginFullName,
-		path:           path,
+		name:           name,
+		args:           args,
 		commitID:       commitID,
 		isWasm:         isWasm,
 		isLocal:        isLocal,
@@ -178,11 +224,15 @@ func (p *plugin) OpaqueID() string {
 	if p.pluginFullName != nil {
 		return p.pluginFullName.String()
 	}
-	return strings.Join(p.path, " ")
+	return p.name + " " + strings.Join(p.args, " ")
+}
+
+func (p *plugin) Name() string {
+	return p.name
 }
 
-func (p *plugin) Path() []string {
-	return p.path
+func (p *plugin) Args() []string {
+	return p.args
 }
 
 func (p *plugin) FullName() bufparse.FullName {
diff --git a/private/bufpkg/bufplugin/plugin_key_provider.go b/private/bufpkg/bufplugin/plugin_key_provider.go
index fcde880280..cef430f5db 100644
--- a/private/bufpkg/bufplugin/plugin_key_provider.go
+++ b/private/bufpkg/bufplugin/plugin_key_provider.go
@@ -16,9 +16,11 @@ package bufplugin
 
 import (
 	"context"
+	"fmt"
 	"io/fs"
 
 	"github.com/bufbuild/buf/private/bufpkg/bufparse"
+	"github.com/bufbuild/buf/private/pkg/slicesext"
 )
 
 var (
@@ -41,6 +43,32 @@ type PluginKeyProvider interface {
 	GetPluginKeysForPluginRefs(context.Context, []bufparse.Ref, DigestType) ([]PluginKey, error)
 }
 
+// NewStaticPluginKeyProvider returns a new PluginKeyProvider for a set of PluginKeys.
+//
+// The set of PluginKeys must be unique by FullName. If there are duplicates,
+// an error will be returned. The Ref will be matched to the PluginKey by FullName.
+// If the Ref is not found in the set of provided keys, an error with
+// fs.ErrNotExist will be returned.
+func NewStaticPluginKeyProvider(pluginKeys []PluginKey) (PluginKeyProvider, error) {
+	if len(pluginKeys) == 0 {
+		return NopPluginKeyProvider, nil
+	}
+	pluginKeysByFullName, err := slicesext.ToUniqueValuesMap(pluginKeys, func(pluginKey PluginKey) string {
+		return pluginKey.FullName().String()
+	})
+	if err != nil {
+		return nil, err
+	}
+	digetType, err := UniqueDigestTypeForPluginKeys(pluginKeys)
+	if err != nil {
+		return nil, err
+	}
+	return staticPluginKeyProvider{
+		pluginKeysByFullName: pluginKeysByFullName,
+		digestType:           digetType,
+	}, nil
+}
+
 // *** PRIVATE ***
 
 type nopPluginKeyProvider struct{}
@@ -52,3 +80,30 @@ func (nopPluginKeyProvider) GetPluginKeysForPluginRefs(
 ) ([]PluginKey, error) {
 	return nil, fs.ErrNotExist
 }
+
+type staticPluginKeyProvider struct {
+	pluginKeysByFullName map[string]PluginKey
+	digestType           DigestType
+}
+
+func (s staticPluginKeyProvider) GetPluginKeysForPluginRefs(
+	_ context.Context,
+	refs []bufparse.Ref,
+	digestType DigestType,
+) ([]PluginKey, error) {
+	if digestType != s.digestType {
+		return nil, fmt.Errorf("expected DigestType %v, got %v", s.digestType, digestType)
+	}
+	pluginKeys := make([]PluginKey, len(refs))
+	for i, ref := range refs {
+		// Only the FullName is used to match the PluginKey. The Ref is not
+		// validated to match the PluginKey as there is not enough information
+		// to do so.
+		pluginKey, ok := s.pluginKeysByFullName[ref.FullName().String()]
+		if !ok {
+			return nil, fs.ErrNotExist
+		}
+		pluginKeys[i] = pluginKey
+	}
+	return pluginKeys, nil
+}
diff --git a/private/pkg/pluginrpcutil/pluginrpcutil.go b/private/pkg/pluginrpcutil/pluginrpcutil.go
index 3164e316fa..326bc87ce8 100644
--- a/private/pkg/pluginrpcutil/pluginrpcutil.go
+++ b/private/pkg/pluginrpcutil/pluginrpcutil.go
@@ -15,18 +15,54 @@
 package pluginrpcutil
 
 import (
+	"fmt"
+	"os"
+
 	"github.com/bufbuild/buf/private/pkg/wasm"
 	"pluginrpc.com/pluginrpc"
 )
 
-// NewRunner returns a new pluginrpc.Runner for the program name.
-func NewRunner(programName string, programArgs ...string) pluginrpc.Runner {
+// NewLocalRunner returns a new pluginrpc.Runner for the local program.
+//
+// The programName is the path or name of the program. Any program args are passed to
+// the program when it is run. The programArgs may be nil.
+func NewLocalRunner(programName string, programArgs ...string) pluginrpc.Runner {
 	return newRunner(programName, programArgs...)
 }
 
-// NewWasmRunner returns a new pluginrpc.Runner for the wasm.Runtime and program name.
+// NewWasmRunner returns a new pluginrpc.Runner for the wasm.Runtime.
+//
+// The getData function should return the Wasm module bytes for the program.
+// The program name is the name of the program. Any program args are passed to
+// the program when it is run. The programArgs may be nil.
+func NewWasmRunner(delegate wasm.Runtime, getData func() ([]byte, error), programName string, programArgs ...string) pluginrpc.Runner {
+	return newWasmRunner(delegate, getData, programName, programArgs...)
+}
+
+// NewLocalWasmRunner returns a new pluginrpc.Runner for the wasm.Runtime.
 //
-// This runner is used for local Wasm plugins. The program name is the path to the Wasm file.
-func NewWasmRunner(delegate wasm.Runtime, programName string, programArgs ...string) pluginrpc.Runner {
-	return newWasmRunner(delegate, programName, programArgs...)
+// The program name is the path to the Wasm file. Any program args are passed to
+// the program when it is run. The programArgs may be nil.
+func NewLocalWasmRunner(delegate wasm.Runtime, programName string, programArgs ...string) pluginrpc.Runner {
+	getData := func() ([]byte, error) {
+		// Find the plugin filePath. We use the same logic as exec.LookPath, but we do
+		// not require the file to be executable. So check the local directory
+		// first before checking the PATH.
+		var filePath string
+		if fileInfo, err := os.Stat(programName); err == nil && !fileInfo.IsDir() {
+			filePath = programName
+		} else {
+			var err error
+			filePath, err = unsafeLookPath(programName)
+			if err != nil {
+				return nil, fmt.Errorf("could not find plugin %q in PATH: %v", programName, err)
+			}
+		}
+		moduleWasm, err := os.ReadFile(filePath)
+		if err != nil {
+			return nil, fmt.Errorf("could not read plugin %q: %v", programName, err)
+		}
+		return moduleWasm, nil
+	}
+	return newWasmRunner(delegate, getData, programName, programArgs...)
 }
diff --git a/private/pkg/pluginrpcutil/wasm_runner.go b/private/pkg/pluginrpcutil/wasm_runner.go
index 0c7640ab41..cba797f8ff 100644
--- a/private/pkg/pluginrpcutil/wasm_runner.go
+++ b/private/pkg/pluginrpcutil/wasm_runner.go
@@ -17,8 +17,6 @@ package pluginrpcutil
 import (
 	"context"
 	"errors"
-	"fmt"
-	"os"
 	"os/exec"
 	"slices"
 	"sync"
@@ -29,6 +27,7 @@ import (
 
 type wasmRunner struct {
 	delegate    wasm.Runtime
+	getData     func() ([]byte, error)
 	programName string
 	programArgs []string
 	// lock protects compiledModule and compiledModuleErr. Store called as
@@ -41,11 +40,13 @@ type wasmRunner struct {
 
 func newWasmRunner(
 	delegate wasm.Runtime,
+	getData func() ([]byte, error),
 	programName string,
 	programArgs ...string,
 ) *wasmRunner {
 	return &wasmRunner{
 		delegate:    delegate,
+		getData:     getData,
 		programName: programName,
 		programArgs: programArgs,
 	}
@@ -79,22 +80,9 @@ func (r *wasmRunner) loadCompiledModuleOnce(ctx context.Context) (wasm.CompiledM
 }
 
 func (r *wasmRunner) loadCompiledModule(ctx context.Context) (wasm.CompiledModule, error) {
-	// Find the plugin path. We use the same logic as exec.LookPath, but we do
-	// not require the file to be executable. So check the local directory
-	// first before checking the PATH.
-	var path string
-	if fileInfo, err := os.Stat(r.programName); err == nil && !fileInfo.IsDir() {
-		path = r.programName
-	} else {
-		var err error
-		path, err = unsafeLookPath(r.programName)
-		if err != nil {
-			return nil, fmt.Errorf("could not find plugin %q in PATH: %v", r.programName, err)
-		}
-	}
-	moduleWasm, err := os.ReadFile(path)
+	moduleWasm, err := r.getData()
 	if err != nil {
-		return nil, fmt.Errorf("could not read plugin %q: %v", r.programName, err)
+		return nil, err
 	}
 	// Compile the module. This CompiledModule is never released, so
 	// subsequent calls to this function will benefit from the cached