feat: complete not-found ConduitError codes (processor instance + plugin registries)

pkg/foundation/cerrors/conduiterr/conduiterr.go

@@ -104,6 +104,9 @@ var (
 	// CodeConnectorPluginNotFound is raised when a referenced connector plugin
 	// cannot be located.
 	CodeConnectorPluginNotFound = Register("connector.plugin_not_found", codes.NotFound)
+	// CodeProcessorPluginNotFound is raised when a referenced processor plugin
+	// cannot be located.
+	CodeProcessorPluginNotFound = Register("processor.plugin_not_found", codes.NotFound)
 )
 
 // Fix is a structured, machine-appliable change that resolves an error. The same

pkg/plugin/connector/standalone/registry.go

@@ -16,6 +16,7 @@ package standalone
 
 import (
 	"context"
+	"fmt"
 	"os"
 	"path"
 	"path/filepath"
@@ -26,6 +27,7 @@ import (
 	"github.com/conduitio/conduit-connector-protocol/pconnector/client"
 	"github.com/conduitio/conduit-connector-protocol/pconnutils"
 	"github.com/conduitio/conduit/pkg/foundation/cerrors"
+	"github.com/conduitio/conduit/pkg/foundation/cerrors/conduiterr"
 	"github.com/conduitio/conduit/pkg/foundation/log"
 	"github.com/conduitio/conduit/pkg/plugin"
 	"github.com/conduitio/conduit/pkg/plugin/connector"
@@ -195,15 +197,29 @@ func (r *Registry) NewDispenser(logger log.CtxLogger, fullName plugin.FullName,
 
 	versionMap, ok := r.plugins[fullName.PluginName()]
 	if !ok {
-		return nil, plugin.ErrPluginNotFound
+		// Invariant: errors.Is(err, plugin.ErrPluginNotFound) still holds — the
+		// sentinel is wrapped, and the ConduitError adds the machine-actionable code.
+		err := conduiterr.Wrap(
+			conduiterr.CodeConnectorPluginNotFound,
+			fmt.Sprintf("standalone connector plugin %q not found", fullName.PluginName()),
+			plugin.ErrPluginNotFound,
+		)
+		err.Suggestion = "check the plugin name and version"
+		return nil, err
 	}
 	bp, ok := versionMap[fullName.PluginVersion()]
 	if !ok {
 		availableVersions := make([]string, 0, len(versionMap))
 		for k := range versionMap {
 			availableVersions = append(availableVersions, k)
 		}
-		return nil, cerrors.Errorf("could not find standalone connector plugin, only found versions %v: %w", availableVersions, plugin.ErrPluginNotFound)
+		err := conduiterr.Wrap(
+			conduiterr.CodeConnectorPluginNotFound,
+			fmt.Sprintf("standalone connector plugin %q not found; available versions: %v", fullName.PluginName(), availableVersions),
+			plugin.ErrPluginNotFound,
+		)
+		err.Suggestion = "check the plugin version, or omit it to use the latest"
+		return nil, err
 	}
 
 	logger = logger.WithComponent("plugin.standalone")

pkg/plugin/processor/builtin/registry.go

@@ -16,12 +16,14 @@ package builtin
 
 import (
 	"context"
+	"fmt"
 	"runtime/debug"
 
 	sdk "github.com/conduitio/conduit-processor-sdk"
 	"github.com/conduitio/conduit-processor-sdk/pprocutils"
 	"github.com/conduitio/conduit-processor-sdk/schema"
 	"github.com/conduitio/conduit/pkg/foundation/cerrors"
+	"github.com/conduitio/conduit/pkg/foundation/cerrors/conduiterr"
 	"github.com/conduitio/conduit/pkg/foundation/ctxutil"
 	"github.com/conduitio/conduit/pkg/foundation/log"
 	"github.com/conduitio/conduit/pkg/plugin"
@@ -190,15 +192,29 @@ func newFullName(pluginName, pluginVersion string) plugin.FullName {
 func (r *Registry) NewProcessor(_ context.Context, fullName plugin.FullName, id string) (sdk.Processor, error) {
 	versionMap, ok := r.plugins[fullName.PluginName()]
 	if !ok {
-		return nil, plugin.ErrPluginNotFound
+		// Invariant: errors.Is(err, plugin.ErrPluginNotFound) still holds — the
+		// sentinel is wrapped, and the ConduitError adds the machine-actionable code.
+		err := conduiterr.Wrap(
+			conduiterr.CodeProcessorPluginNotFound,
+			fmt.Sprintf("builtin processor plugin %q not found", fullName.PluginName()),
+			plugin.ErrPluginNotFound,
+		)
+		err.Suggestion = "check the plugin name and version"
+		return nil, err
 	}
 	b, ok := versionMap[fullName.PluginVersion()]
 	if !ok {
 		availableVersions := make([]string, 0, len(versionMap))
 		for k := range versionMap {
 			availableVersions = append(availableVersions, k)
 		}
-		return nil, cerrors.Errorf("could not find builtin plugin %q, only found versions %v: %w", fullName, availableVersions, plugin.ErrPluginNotFound)
+		err := conduiterr.Wrap(
+			conduiterr.CodeProcessorPluginNotFound,
+			fmt.Sprintf("builtin processor plugin %q not found; available versions: %v", fullName, availableVersions),
+			plugin.ErrPluginNotFound,
+		)
+		err.Suggestion = "check the plugin version, or omit it to use the latest"
+		return nil, err
 	}
 
 	p := b.constructor(r.logger)

pkg/plugin/processor/standalone/registry.go

@@ -25,6 +25,7 @@ import (
 	sdk "github.com/conduitio/conduit-processor-sdk"
 	"github.com/conduitio/conduit-processor-sdk/pprocutils"
 	"github.com/conduitio/conduit/pkg/foundation/cerrors"
+	"github.com/conduitio/conduit/pkg/foundation/cerrors/conduiterr"
 	"github.com/conduitio/conduit/pkg/foundation/log"
 	"github.com/conduitio/conduit/pkg/plugin"
 	"github.com/stealthrocket/wazergo"
@@ -125,15 +126,29 @@ func (r *Registry) NewProcessor(ctx context.Context, fullName plugin.FullName, i
 
 	versions, ok := r.plugins[fullName.PluginName()]
 	if !ok {
-		return nil, plugin.ErrPluginNotFound
+		// Invariant: errors.Is(err, plugin.ErrPluginNotFound) still holds — the
+		// sentinel is wrapped, and the ConduitError adds the machine-actionable code.
+		err := conduiterr.Wrap(
+			conduiterr.CodeProcessorPluginNotFound,
+			fmt.Sprintf("standalone processor plugin %q not found", fullName.PluginName()),
+			plugin.ErrPluginNotFound,
+		)
+		err.Suggestion = "check the plugin name and version"
+		return nil, err
 	}
 	bp, ok := versions[fullName.PluginVersion()]
 	if !ok {
 		availableVersions := make([]string, 0, len(versions))
 		for k := range versions {
 			availableVersions = append(availableVersions, k)
 		}
-		return nil, cerrors.Errorf("could not find standalone processor plugin, only found versions %v: %w", availableVersions, plugin.ErrPluginNotFound)
+		err := conduiterr.Wrap(
+			conduiterr.CodeProcessorPluginNotFound,
+			fmt.Sprintf("standalone processor plugin %q not found; available versions: %v", fullName.PluginName(), availableVersions),
+			plugin.ErrPluginNotFound,
+		)
+		err.Suggestion = "check the plugin version, or omit it to use the latest"
+		return nil, err
 	}
 
 	p, err := newWASMProcessor(ctx, r.runtime, bp.module, r.hostModule, r.schemaService, id, r.logger)

pkg/processor/codes.go

@@ -0,0 +1,29 @@
+// Copyright © 2026 Meroxa, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package processor
+
+import (
+	"github.com/conduitio/conduit/pkg/foundation/cerrors/conduiterr"
+	"google.golang.org/grpc/codes"
+)
+
+// Processor error codes. Every error carries one of these codes plus a
+// suggested fix, so an API, MCP, or UI consumer knows what happened without
+// parsing message text.
+var (
+	// CodeProcessorNotFound is raised when a referenced processor instance
+	// cannot be located.
+	CodeProcessorNotFound = conduiterr.Register("processor.instance_not_found", codes.NotFound)
+)

pkg/processor/service.go

@@ -18,11 +18,13 @@ package processor
 
 import (
 	"context"
+	"fmt"
 	"time"
 
 	"github.com/conduitio/conduit-commons/database"
 	sdk "github.com/conduitio/conduit-processor-sdk"
 	"github.com/conduitio/conduit/pkg/foundation/cerrors"
+	"github.com/conduitio/conduit/pkg/foundation/cerrors/conduiterr"
 	"github.com/conduitio/conduit/pkg/foundation/log"
 	"github.com/conduitio/conduit/pkg/foundation/metrics/measure"
 )
@@ -86,7 +88,15 @@ func (s *Service) List(_ context.Context) map[string]*Instance {
 func (s *Service) Get(_ context.Context, id string) (*Instance, error) {
 	ins, ok := s.instances[id]
 	if !ok {
-		return nil, cerrors.Errorf("%w (ID: %s)", ErrInstanceNotFound, id)
+		// Invariant: errors.Is(err, ErrInstanceNotFound) still holds — the sentinel
+		// is wrapped, and the ConduitError adds the machine-actionable code.
+		err := conduiterr.Wrap(
+			CodeProcessorNotFound,
+			fmt.Sprintf("processor %q not found", id),
+			ErrInstanceNotFound,
+		)
+		err.Suggestion = "check the processor ID in your pipeline configuration"
+		return nil, err
 	}
 	return ins, nil
 }

pkg/processor/service_test.go

@@ -23,6 +23,7 @@ import (
 	dbmock "github.com/conduitio/conduit-commons/database/mock"
 	sdk "github.com/conduitio/conduit-processor-sdk"
 	"github.com/conduitio/conduit/pkg/foundation/cerrors"
+	"github.com/conduitio/conduit/pkg/foundation/cerrors/conduiterr"
 	"github.com/conduitio/conduit/pkg/foundation/log"
 	"github.com/conduitio/conduit/pkg/plugin"
 	proc_plugin "github.com/conduitio/conduit/pkg/plugin/processor"
@@ -328,7 +329,11 @@ func TestService_Get_Fail(t *testing.T) {
 	service := NewService(log.Nop(), db, &proc_plugin.PluginService{})
 
 	got, err := service.Get(ctx, "non-existent processor")
-	is.True(cerrors.Is(err, ErrInstanceNotFound)) // expected instance not found error
+	is.True(cerrors.Is(err, ErrInstanceNotFound)) // sentinel still in the chain
+	ce, ok := conduiterr.Get(err)
+	is.True(ok) // now also carries a machine-actionable ConduitError code
+	is.Equal(ce.Code.Reason(), CodeProcessorNotFound.Reason())
+	is.True(ce.Suggestion != "") // with a suggested fix
 	is.Equal(got, nil)
 }