feat(plugin): emit connector.plugin_not_found ConduitError code

pkg/http/api/status/status.go

@@ -17,6 +17,7 @@ package status
 import (
 	"github.com/conduitio/conduit/pkg/connector"
 	"github.com/conduitio/conduit/pkg/foundation/cerrors"
+	"github.com/conduitio/conduit/pkg/foundation/cerrors/conduiterr"
 	"github.com/conduitio/conduit/pkg/orchestrator"
 	"github.com/conduitio/conduit/pkg/pipeline"
 	conn_plugin "github.com/conduitio/conduit/pkg/plugin/connector"
@@ -25,7 +26,24 @@ import (
 	grpcstatus "google.golang.org/grpc/status"
 )
 
+// conduitErrorStatus returns the structured gRPC status for a ConduitError in err's
+// chain — its gRPC category plus a google.rpc.ErrorInfo detail carrying the stable
+// reason and any configPath/suggestion/fix — and true if one was found. This is the
+// single point where the ConduitError model reaches the API; boundaries that have
+// not yet been migrated fall through to the sentinel-based code mapping below,
+// unchanged.
+func conduitErrorStatus(err error) (error, bool) {
+	if ce, ok := conduiterr.Get(err); ok {
+		return conduiterr.ToStatus(ce).Err(), true
+	}
+	return nil, false
+}
+
 func PipelineError(err error) error {
+	if s, ok := conduitErrorStatus(err); ok {
+		return s
+	}
+
 	var code codes.Code
 
 	switch {
@@ -41,6 +59,10 @@ func PipelineError(err error) error {
 }
 
 func ConnectorError(err error) error {
+	if s, ok := conduitErrorStatus(err); ok {
+		return s
+	}
+
 	var code codes.Code
 
 	switch {
@@ -56,6 +78,10 @@ func ConnectorError(err error) error {
 }
 
 func ProcessorError(err error) error {
+	if s, ok := conduitErrorStatus(err); ok {
+		return s
+	}
+
 	var code codes.Code
 
 	switch {
@@ -71,6 +97,9 @@ func ProcessorError(err error) error {
 }
 
 func PluginError(err error) error {
+	if s, ok := conduitErrorStatus(err); ok {
+		return s
+	}
 	return grpcstatus.Error(codeFromError(err), err.Error())
 }
 

pkg/http/api/status/status_test.go

@@ -19,10 +19,12 @@ import (
 
 	"github.com/conduitio/conduit/pkg/connector"
 	"github.com/conduitio/conduit/pkg/foundation/cerrors"
+	"github.com/conduitio/conduit/pkg/foundation/cerrors/conduiterr"
 	"github.com/conduitio/conduit/pkg/orchestrator"
 	"github.com/conduitio/conduit/pkg/pipeline"
 	"github.com/conduitio/conduit/pkg/processor"
 	"github.com/matryer/is"
+	"google.golang.org/genproto/googleapis/rpc/errdetails"
 	"google.golang.org/grpc/codes"
 	grpcstatus "google.golang.org/grpc/status"
 )
@@ -126,6 +128,45 @@ func TestProcessorError(t *testing.T) {
 	}
 }
 
+// TestConduitErrorFlowsThroughBoundary proves the ConduitError foundation reaches
+// the API: a ConduitError (even wrapped) yields a gRPC status with the code's
+// category and an additive google.rpc.ErrorInfo detail carrying the stable reason,
+// configPath, and suggestion.
+func TestConduitErrorFlowsThroughBoundary(t *testing.T) {
+	is := is.New(t)
+
+	ce := conduiterr.New(conduiterr.CodeConnectorPluginNotFound, "connector plugin not found")
+	ce.ConfigPath = "/connectors/0/plugin"
+	ce.Suggestion = "run `conduit connectors install <name>`"
+
+	// wrapped, to prove errors.As finds it through the chain
+	got := PipelineError(cerrors.Errorf("provisioning failed: %w", ce))
+
+	st, ok := grpcstatus.FromError(got)
+	is.True(ok)
+	is.Equal(st.Code(), codes.NotFound) // the code's gRPC category
+
+	var info *errdetails.ErrorInfo
+	for _, d := range st.Details() {
+		if ei, ok := d.(*errdetails.ErrorInfo); ok {
+			info = ei
+		}
+	}
+	is.True(info != nil) // the structured detail is present
+	is.Equal(info.GetReason(), conduiterr.CodeConnectorPluginNotFound.Reason())
+	is.Equal(info.GetDomain(), "conduit")
+	is.Equal(info.GetMetadata()["configPath"], "/connectors/0/plugin")
+	is.Equal(info.GetMetadata()["suggestion"], "run `conduit connectors install <name>`")
+}
+
+// TestNonConduitErrorUnchanged confirms the wiring is additive: a plain sentinel
+// error still maps exactly as before, with no detail.
+func TestNonConduitErrorUnchanged(t *testing.T) {
+	is := is.New(t)
+	got := PipelineError(pipeline.ErrInstanceNotFound)
+	is.Equal(grpcstatus.Error(codes.NotFound, pipeline.ErrInstanceNotFound.Error()), got)
+}
+
 func TestCodeFromError(t *testing.T) {
 	type args struct {
 		err error

pkg/plugin/connector/builtin/registry.go

@@ -16,6 +16,7 @@ package builtin
 
 import (
 	"context"
+	"fmt"
 	"runtime/debug"
 
 	file "github.com/conduitio/conduit-connector-file"
@@ -28,6 +29,7 @@ import (
 	sdk "github.com/conduitio/conduit-connector-sdk"
 	"github.com/conduitio/conduit-connector-sdk/schema"
 	"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"
@@ -190,15 +192,29 @@ func newFullName(pluginName, pluginVersion string) plugin.FullName {
 func (r *Registry) NewDispenser(logger log.CtxLogger, fullName plugin.FullName, cfg pconnector.PluginConfig) (connector.Dispenser, 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.CodeConnectorPluginNotFound,
+			fmt.Sprintf("builtin connector plugin %q not found", fullName.PluginName()),
+			plugin.ErrPluginNotFound,
+		)
+		err.Suggestion = "run `conduit connectors list` to see available built-in connectors"
+		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.CodeConnectorPluginNotFound,
+			fmt.Sprintf("builtin connector 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
 	}
 
 	return b.dispenserFactory(fullName, cfg, logger), nil

pkg/plugin/connector/builtin/registry_test.go

@@ -20,6 +20,7 @@ import (
 
 	"github.com/conduitio/conduit-connector-protocol/pconnector"
 	"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/google/go-cmp/cmp"
@@ -98,7 +99,11 @@ func TestRegistry_NewDispenser_PluginNotFound(t *testing.T) {
 
 			dispenser, err := underTest.NewDispenser(log.Nop(), plugin.FullName(tc.pluginName), pconnector.PluginConfig{})
 			if tc.wantErr {
-				is.True(cerrors.Is(err, plugin.ErrPluginNotFound))
+				is.True(cerrors.Is(err, plugin.ErrPluginNotFound)) // 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(), conduiterr.CodeConnectorPluginNotFound.Reason())
+				is.True(ce.Suggestion != "") // with a suggested fix
 				is.True(dispenser == nil)
 				return
 			}