feat(config): emit ConduitError codes + configPath for pipeline validation

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
 			}

pkg/provisioning/config/codes.go

@@ -0,0 +1,39 @@
+// 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 config
+
+import (
+	"github.com/conduitio/conduit/pkg/foundation/cerrors/conduiterr"
+	"google.golang.org/grpc/codes"
+)
+
+// Pipeline-config validation error codes. Every validation error carries one of
+// these codes plus a JSON-pointer configPath to the offending field, so an API,
+// MCP, or UI consumer knows exactly what to fix and where.
+var (
+	CodeFieldRequired = conduiterr.Register("config.field_required", codes.InvalidArgument)
+	CodeFieldInvalid  = conduiterr.Register("config.field_invalid", codes.InvalidArgument)
+	CodeFieldTooLong  = conduiterr.Register("config.field_too_long", codes.InvalidArgument)
+	CodeIDDuplicate   = conduiterr.Register("config.id_duplicate", codes.InvalidArgument)
+)
+
+// fieldError builds a ConduitError carrying a machine-actionable code and the
+// JSON-pointer path to the offending field. The sentinel is kept in the chain so
+// existing errors.Is checks (ErrMandatoryField, ErrInvalidField, …) still hold.
+func fieldError(code conduiterr.Code, configPath, msg string, sentinel error) error {
+	e := conduiterr.Wrap(code, msg, sentinel)
+	e.ConfigPath = configPath
+	return e
+}

pkg/provisioning/config/validate.go

@@ -15,6 +15,8 @@
 package config
 
 import (
+	"fmt"
+
 	"github.com/conduitio/conduit/pkg/connector"
 	"github.com/conduitio/conduit/pkg/foundation/cerrors"
 	"github.com/conduitio/conduit/pkg/pipeline"
@@ -27,84 +29,89 @@ const (
 	TypeDestination = "destination"
 )
 
-// Validate validates config field values for a pipeline
+// Validate validates config field values for a pipeline. Each error carries a
+// ConduitError code and a JSON-pointer configPath to the offending field; the
+// original sentinel stays in the chain for backward-compatible errors.Is checks.
 func Validate(cfg Pipeline) error {
 	var errs []error
 	if cfg.ID == "" {
-		errs = append(errs, cerrors.Errorf(`id is mandatory: %w`, ErrMandatoryField))
+		errs = append(errs, fieldError(CodeFieldRequired, "/id", "id is mandatory", ErrMandatoryField))
 	}
 	if len(cfg.ID) > pipeline.IDLengthLimit {
-		errs = append(errs, pipeline.ErrIDOverLimit)
+		errs = append(errs, fieldError(CodeFieldTooLong, "/id", pipeline.ErrIDOverLimit.Error(), pipeline.ErrIDOverLimit))
 	}
 	if len(cfg.Name) > pipeline.NameLengthLimit {
-		errs = append(errs, pipeline.ErrNameOverLimit)
+		errs = append(errs, fieldError(CodeFieldTooLong, "/name", pipeline.ErrNameOverLimit.Error(), pipeline.ErrNameOverLimit))
 	}
 	if len(cfg.Description) > pipeline.DescriptionLengthLimit {
-		errs = append(errs, pipeline.ErrDescriptionOverLimit)
+		errs = append(errs, fieldError(CodeFieldTooLong, "/description", pipeline.ErrDescriptionOverLimit.Error(), pipeline.ErrDescriptionOverLimit))
 	}
 	if cfg.Status != StatusRunning && cfg.Status != StatusStopped {
-		errs = append(errs, cerrors.Errorf(`"status" is invalid: %w`, ErrInvalidField))
+		errs = append(errs, fieldError(CodeFieldInvalid, "/status", `"status" is invalid`, ErrInvalidField))
 	}
 
 	errs = append(errs, validateConnectors(cfg.Connectors)...)
-	errs = append(errs, validateProcessors(cfg.Processors)...)
+	errs = append(errs, validateProcessors(cfg.Processors, "/processors")...)
 
 	return cerrors.Join(errs...)
 }
 
-// validateConnectors validates config field values for connectors
+// validateConnectors validates config field values for connectors. pathPrefix is
+// the JSON-pointer to the connectors slice ("/connectors").
 func validateConnectors(mp []Connector) []error {
 	var errs []error
 	ids := make(map[string]bool)
 	for i, cfg := range mp {
+		path := fmt.Sprintf("/connectors/%d", i)
 		if cfg.ID == "" {
-			errs = append(errs, cerrors.Errorf("connector %d: id is mandatory: %w", i+1, ErrMandatoryField))
+			errs = append(errs, fieldError(CodeFieldRequired, path+"/id", fmt.Sprintf("connector %d: id is mandatory", i+1), ErrMandatoryField))
 		}
 		if len(cfg.ID) > connector.IDLengthLimit {
-			errs = append(errs, cerrors.Errorf("connector %q: %w", cfg.ID, connector.ErrIDOverLimit))
+			errs = append(errs, fieldError(CodeFieldTooLong, path+"/id", fmt.Sprintf("connector %q: %s", cfg.ID, connector.ErrIDOverLimit), connector.ErrIDOverLimit))
 		}
 		if len(cfg.Name) > connector.NameLengthLimit {
-			errs = append(errs, cerrors.Errorf("connector %q: %w", cfg.ID, connector.ErrNameOverLimit))
+			errs = append(errs, fieldError(CodeFieldTooLong, path+"/name", fmt.Sprintf("connector %q: %s", cfg.ID, connector.ErrNameOverLimit), connector.ErrNameOverLimit))
 		}
 		if cfg.Plugin == "" {
-			errs = append(errs, cerrors.Errorf(`connector %q: "plugin" is mandatory: %w`, cfg.ID, ErrMandatoryField))
+			errs = append(errs, fieldError(CodeFieldRequired, path+"/plugin", fmt.Sprintf(`connector %q: "plugin" is mandatory`, cfg.ID), ErrMandatoryField))
 		}
 		if cfg.Type == "" {
-			errs = append(errs, cerrors.Errorf(`connector %q: "type" is mandatory: %w`, cfg.ID, ErrMandatoryField))
+			errs = append(errs, fieldError(CodeFieldRequired, path+"/type", fmt.Sprintf(`connector %q: "type" is mandatory`, cfg.ID), ErrMandatoryField))
 		}
 		if cfg.Type != "" && cfg.Type != TypeSource && cfg.Type != TypeDestination {
-			errs = append(errs, cerrors.Errorf(`connector %q: "type" is invalid: %w`, cfg.ID, ErrInvalidField))
+			errs = append(errs, fieldError(CodeFieldInvalid, path+"/type", fmt.Sprintf(`connector %q: "type" is invalid`, cfg.ID), ErrInvalidField))
 		}
 
-		pErrs := validateProcessors(cfg.Processors)
-		for _, pErr := range pErrs {
-			errs = append(errs, cerrors.Errorf("connector %q: %w", cfg.ID, pErr))
-		}
+		// nested processor errors already carry their own /connectors/i/processors/j path
+		errs = append(errs, validateProcessors(cfg.Processors, path+"/processors")...)
 
 		if ids[cfg.ID] {
-			errs = append(errs, cerrors.Errorf(`connector %q: "id" must be unique: %w`, cfg.ID, ErrDuplicateID))
+			errs = append(errs, fieldError(CodeIDDuplicate, path+"/id", fmt.Sprintf(`connector %q: "id" must be unique`, cfg.ID), ErrDuplicateID))
 		}
 		ids[cfg.ID] = true
 	}
 	return errs
 }
 
-// validateProcessorsConfig validates config field values for processors
-func validateProcessors(mp []Processor) []error {
+// validateProcessors validates config field values for processors. pathPrefix is
+// the JSON-pointer to the processors slice (e.g. "/processors" or
+// "/connectors/0/processors").
+func validateProcessors(mp []Processor, pathPrefix string) []error {
 	var errs []error
 	ids := make(map[string]bool)
 	for i, cfg := range mp {
+		path := fmt.Sprintf("%s/%d", pathPrefix, i)
 		if cfg.ID == "" {
-			errs = append(errs, cerrors.Errorf("processor %d: id is mandatory: %w", i+1, ErrMandatoryField))
+			errs = append(errs, fieldError(CodeFieldRequired, path+"/id", fmt.Sprintf("processor %d: id is mandatory", i+1), ErrMandatoryField))
 		}
 		if cfg.Plugin == "" {
-			errs = append(errs, cerrors.Errorf(`processor %q: "plugin" is mandatory: %w`, cfg.ID, ErrMandatoryField))
+			errs = append(errs, fieldError(CodeFieldRequired, path+"/plugin", fmt.Sprintf(`processor %q: "plugin" is mandatory`, cfg.ID), ErrMandatoryField))
 		}
 		if cfg.Workers < 0 {
-			errs = append(errs, cerrors.Errorf(`processor %q: "workers" can't be negative: %w`, cfg.ID, ErrInvalidField))
+			errs = append(errs, fieldError(CodeFieldInvalid, path+"/workers", fmt.Sprintf(`processor %q: "workers" can't be negative`, cfg.ID), ErrInvalidField))
 		}
 		if ids[cfg.ID] {
-			errs = append(errs, cerrors.Errorf(`processor %q: "id" must be unique: %w`, cfg.ID, ErrDuplicateID))
+			errs = append(errs, fieldError(CodeIDDuplicate, path+"/id", fmt.Sprintf(`processor %q: "id" must be unique`, cfg.ID), ErrDuplicateID))
 		}
 		ids[cfg.ID] = true
 	}