DON-Level Metrics: Emit classified workflow errors

.changeset/classified-execution-status.md

@@ -0,0 +1,5 @@
+---
+"chainlink": minor
+---
+
+#added The v2 workflow engine now emits `ClassifiedExecutionStatus` on `WorkflowExecutionFinished` events, distinguishing failures caused by the user's workflow (`USER_ERROR`) from platform/infrastructure failures (`SYSTEM_ERROR`). The v1 engine is unaffected.

core/scripts/go.mod

@@ -513,7 +513,7 @@ require (
 	github.com/smartcontractkit/chainlink-protos/ring/go v0.0.0-20260331131315-f08a616d8dcd // indirect
 	github.com/smartcontractkit/chainlink-protos/storage-service v0.3.0 // indirect
 	github.com/smartcontractkit/chainlink-protos/svr v1.3.0 // indirect
-	github.com/smartcontractkit/chainlink-protos/workflows/go v0.0.0-20260630073003-fb8da7229930 // indirect
+	github.com/smartcontractkit/chainlink-protos/workflows/go v0.0.0-20260716141634-c0cc05ed05d8 // indirect
 	github.com/smartcontractkit/chainlink-solana/contracts v0.0.0-20260513123719-d347eaf314e1 // indirect
 	github.com/smartcontractkit/chainlink-stellar v0.0.3-0.20260714222251-887c274c6b64 // indirect
 	github.com/smartcontractkit/chainlink-stellar/bindings v0.0.0-20260714222251-887c274c6b64 // indirect

core/services/workflows/engine.go

@@ -717,7 +717,8 @@ func (e *Engine) finishExecution(ctx context.Context, cma custmsg.MessageEmitter
 
 	logCustMsg(ctx, cma, fmt.Sprintf("execution duration: %d (seconds)", executionDuration), l)
 	l.Infof("execution duration: %d (seconds)", executionDuration)
-	err = events.EmitExecutionFinishedEvent(ctx, cma.Labels(), status, executionID, nil, l)
+	// v1 engine does not classify user vs system failures.
+	err = events.EmitExecutionFinishedEvent(ctx, cma.Labels(), status, executionID, nil, events.ErrorClassificationUnspecified, l)
 	if err != nil {
 		e.logger.Errorf("failed to emit execution finished event: %+v", err)
 	}

core/services/workflows/events/emit.go

@@ -12,6 +12,7 @@ import (
 	"google.golang.org/protobuf/proto"
 
 	"github.com/smartcontractkit/chainlink-common/pkg/beholder"
+	caperrors "github.com/smartcontractkit/chainlink-common/pkg/capabilities/errors"
 	"github.com/smartcontractkit/chainlink-common/pkg/durableemitter"
 	"github.com/smartcontractkit/chainlink-common/pkg/logger"
 	"github.com/smartcontractkit/chainlink-common/pkg/types"
@@ -215,7 +216,62 @@ func EmitExecutionProfile(
 	return profile, emitProtoMessage(ctx, profile)
 }
 
-func EmitExecutionFinishedEvent(ctx context.Context, labels map[string]string, status string, executionID string, execErr error, lggr logger.Logger) error {
+// ErrorClassification attributes a failed workflow execution to its root cause:
+// the user's workflow (code, config, or a returned error) vs the platform
+// (timeouts, unavailable capabilities, internal engine errors).
+type ErrorClassification int
+
+const (
+	// ErrorClassificationUnspecified means the cause was not determined (e.g. the
+	// legacy v1 engine, which does not classify). Emitted as UNSPECIFIED.
+	ErrorClassificationUnspecified ErrorClassification = iota
+	// ErrorClassificationUser attributes the failure to the user's workflow.
+	ErrorClassificationUser
+	// ErrorClassificationSystem attributes the failure to the platform.
+	ErrorClassificationSystem
+)
+
+// ClassifyError resolves an execution error to a user/system classification.
+// A caperrors.Error carrying an explicit origin wins (so user-origin errors that
+// propagate up from a capability or the guest are attributed correctly); other
+// errors fall back to the caller-supplied classification.
+func ClassifyError(execErr error, fallback ErrorClassification) ErrorClassification {
+	if execErr == nil {
+		return ErrorClassificationUnspecified
+	}
+	var capErr caperrors.Error
+	if errors.As(execErr, &capErr) {
+		switch capErr.Origin() {
+		case caperrors.OriginUser:
+			return ErrorClassificationUser
+		case caperrors.OriginSystem:
+			return ErrorClassificationSystem
+		}
+	}
+	return fallback
+}
+
+// classifiedExecutionStatus folds the terminal status and the error
+// classification into the v2 ClassifiedExecutionStatus enum.
+func classifiedExecutionStatus(status eventsv2.ExecutionStatus, errClass ErrorClassification) eventsv2.ClassifiedExecutionStatus {
+	switch status {
+	case eventsv2.ExecutionStatus_EXECUTION_STATUS_SUCCEEDED:
+		return eventsv2.ClassifiedExecutionStatus_CLASSIFIED_EXECUTION_STATUS_SUCCEEDED
+	case eventsv2.ExecutionStatus_EXECUTION_STATUS_FAILED:
+		switch errClass {
+		case ErrorClassificationUser:
+			return eventsv2.ClassifiedExecutionStatus_CLASSIFIED_EXECUTION_STATUS_USER_ERROR
+		case ErrorClassificationSystem:
+			return eventsv2.ClassifiedExecutionStatus_CLASSIFIED_EXECUTION_STATUS_SYSTEM_ERROR
+		default:
+			return eventsv2.ClassifiedExecutionStatus_CLASSIFIED_EXECUTION_STATUS_UNSPECIFIED
+		}
+	default:
+		return eventsv2.ClassifiedExecutionStatus_CLASSIFIED_EXECUTION_STATUS_UNSPECIFIED
+	}
+}
+
+func EmitExecutionFinishedEvent(ctx context.Context, labels map[string]string, status string, executionID string, execErr error, errClass ErrorClassification, lggr logger.Logger) error {
 	metadata := buildWorkflowMetadata(labels, executionID)
 
 	event := &events.WorkflowExecutionFinished{
@@ -251,6 +307,7 @@ func EmitExecutionFinishedEvent(ctx context.Context, labels map[string]string, s
 		Timestamp:           time.Now().Format(time.RFC3339),
 		Status:              executionStatus,
 		Error:               errMsg,
+		ClassifiedStatus:    classifiedExecutionStatus(executionStatus, errClass),
 	}
 
 	// Emit both v1 and v2 events

core/services/workflows/events/emit_test.go

@@ -46,7 +46,7 @@ func TestEmit(t *testing.T) { //nolint:paralleltest // beholdertest.NewObserver
 	})
 
 	t.Run(events.WorkflowExecutionFinished, func(t *testing.T) { //nolint:paralleltest // shares beholder observer
-		require.NoError(t, events.EmitExecutionFinishedEvent(t.Context(), labels, "status", executionID, nil, nil))
+		require.NoError(t, events.EmitExecutionFinishedEvent(t.Context(), labels, "status", executionID, nil, events.ErrorClassificationUnspecified, nil))
 		require.Len(t, labels, 1)
 
 		msgs := beholderObserver.Messages(t, "beholder_entity", "workflows.v1."+events.WorkflowExecutionFinished)
@@ -60,7 +60,7 @@ func TestEmit(t *testing.T) { //nolint:paralleltest // beholdertest.NewObserver
 
 	t.Run(events.WorkflowExecutionFinished+"_with_error", func(t *testing.T) { //nolint:paralleltest // shares beholder observer
 		testErr := errors.New("something went wrong")
-		require.NoError(t, events.EmitExecutionFinishedEvent(t.Context(), labels, "errored", executionID, testErr, nil))
+		require.NoError(t, events.EmitExecutionFinishedEvent(t.Context(), labels, "errored", executionID, testErr, events.ErrorClassificationSystem, nil))
 
 		v2Msgs := beholderObserver.Messages(t, "beholder_entity", "workflows.v2."+events.WorkflowExecutionFinished)
 		require.NotEmpty(t, v2Msgs)
@@ -70,6 +70,7 @@ func TestEmit(t *testing.T) { //nolint:paralleltest // beholdertest.NewObserver
 		require.NoError(t, proto.Unmarshal(v2Msgs[len(v2Msgs)-1].Body, &v2Event))
 		assert.Equal(t, "something went wrong", v2Event.Error)
 		assert.Equal(t, eventsv2.ExecutionStatus_EXECUTION_STATUS_FAILED, v2Event.Status)
+		assert.Equal(t, eventsv2.ClassifiedExecutionStatus_CLASSIFIED_EXECUTION_STATUS_SYSTEM_ERROR, v2Event.ClassifiedStatus)
 	})
 
 	t.Run(events.CapabilityExecutionStarted, func(t *testing.T) { //nolint:paralleltest // shares beholder observer

core/services/workflows/v2/engine.go

@@ -876,11 +876,13 @@ func (e *Engine) startExecution(ctx context.Context, wrappedTriggerEvent enqueue
 	}
 	e.metrics.With("workflowID", e.cfg.WorkflowID, "workflowName", e.cfg.WorkflowName.String()).IncrementWorkflowExecutionStartedCounter(ctx)
 
-	// Track execution error for deferred event emission
+	// Track execution error (and its user/system classification) for deferred
+	// event emission. Set alongside executionStatus at each outcome site below.
 	var execErr error
+	execErrClass := events.ErrorClassificationUnspecified
 	var execHelper *ExecutionHelper
 	defer func() {
-		_ = events.EmitExecutionFinishedEvent(ctx, loggerLabels, executionStatus, executionID, execErr, lggr)
+		_ = events.EmitExecutionFinishedEvent(ctx, loggerLabels, executionStatus, executionID, execErr, execErrClass, lggr)
 		if execHelper != nil {
 			endTime := e.cfg.Clock.Now()
 			profile, emitErr := events.EmitExecutionProfile(
@@ -932,13 +934,15 @@ func (e *Engine) startExecution(ctx context.Context, wrappedTriggerEvent enqueue
 		lggr.Errorw("Failed to get execution response size limit", "err", err)
 		executionStatus = store.StatusErrored
 		execErr = err
+		execErrClass = events.ErrorClassificationSystem
 		triggerDrop(monitoring.TriggerDropReasonExecutionResponseLimitReadFailed)
 		return
 	}
 	if moduleExecuteMaxResponseSizeBytes < 0 {
 		execErr = fmt.Errorf("invalid moduleExecuteMaxResponseSizeBytes; must not be negative: %d", moduleExecuteMaxResponseSizeBytes)
 		lggr.Errorw(execErr.Error())
 		executionStatus = store.StatusErrored
+		execErrClass = events.ErrorClassificationSystem
 		triggerDrop(monitoring.TriggerDropReasonExecutionResponseLimitInvalid)
 		return
 	}
@@ -1010,6 +1014,10 @@ func (e *Engine) startExecution(ctx context.Context, wrappedTriggerEvent enqueue
 
 	if execErr != nil {
 		executionStatus = store.StatusErrored
+		// Module/host and timeout failures are platform errors by default, but a
+		// user-origin caperrors.Error propagating from a capability or the guest
+		// is attributed to the user.
+		execErrClass = events.ClassifyError(execErr, events.ErrorClassificationSystem)
 		if errors.Is(execErr, context.DeadlineExceeded) {
 			executionStatus = store.StatusTimeout
 			e.metrics.UpdateWorkflowTimeoutDurationHistogram(ctx, int64(executionDuration.Seconds()))
@@ -1028,6 +1036,8 @@ func (e *Engine) startExecution(ctx context.Context, wrappedTriggerEvent enqueue
 	if len(result.GetError()) > 0 {
 		executionStatus = store.StatusErrored
 		execErr = errors.New(result.GetError())
+		// The user's workflow ran and returned an error: a user failure.
+		execErrClass = events.ErrorClassificationUser
 		e.metrics.UpdateWorkflowErrorDurationHistogram(ctx, int64(executionDuration.Seconds()))
 		e.metrics.With("workflowID", e.cfg.WorkflowID, "workflowName", e.cfg.WorkflowName.String()).IncrementWorkflowExecutionFailedCounter(ctx)
 		e.metrics.IncrementWorkflowExecutionFinishedCounter(ctx, executionStatus)

deployment/go.mod

@@ -446,7 +446,7 @@ require (
 	github.com/smartcontractkit/chainlink-protos/ring/go v0.0.0-20260331131315-f08a616d8dcd // indirect
 	github.com/smartcontractkit/chainlink-protos/storage-service v0.3.0 // indirect
 	github.com/smartcontractkit/chainlink-protos/svr v1.3.0 // indirect
-	github.com/smartcontractkit/chainlink-protos/workflows/go v0.0.0-20260630073003-fb8da7229930 // indirect
+	github.com/smartcontractkit/chainlink-protos/workflows/go v0.0.0-20260716141634-c0cc05ed05d8 // indirect
 	github.com/smartcontractkit/chainlink-sui/codec v0.0.0-20260714120433-7667cad5ff5c // indirect
 	github.com/smartcontractkit/chainlink-testing-framework/parrot v0.6.2 // indirect
 	github.com/smartcontractkit/chainlink-testing-framework/seth v1.51.5 // indirect

go.mod

@@ -104,7 +104,7 @@ require (
 	github.com/smartcontractkit/chainlink-protos/orchestrator v0.11.0
 	github.com/smartcontractkit/chainlink-protos/ring/go v0.0.0-20260331131315-f08a616d8dcd
 	github.com/smartcontractkit/chainlink-protos/storage-service v0.3.0
-	github.com/smartcontractkit/chainlink-protos/workflows/go v0.0.0-20260630073003-fb8da7229930
+	github.com/smartcontractkit/chainlink-protos/workflows/go v0.0.0-20260716141634-c0cc05ed05d8
 	github.com/smartcontractkit/chainlink-sui/codec v0.0.0-20260714120433-7667cad5ff5c
 	github.com/smartcontractkit/chainlink-ton/cciplib v0.1.1-0.20260715200135-39296e69ee4e
 	github.com/smartcontractkit/cre-sdk-go v1.9.0-capdev.1.0.20260625132924-dcceeb57cf3c

integration-tests/go.mod

@@ -429,7 +429,7 @@ require (
 	github.com/smartcontractkit/chainlink-protos/ring/go v0.0.0-20260331131315-f08a616d8dcd // indirect
 	github.com/smartcontractkit/chainlink-protos/storage-service v0.3.0 // indirect
 	github.com/smartcontractkit/chainlink-protos/svr v1.3.0 // indirect
-	github.com/smartcontractkit/chainlink-protos/workflows/go v0.0.0-20260630073003-fb8da7229930 // indirect
+	github.com/smartcontractkit/chainlink-protos/workflows/go v0.0.0-20260716141634-c0cc05ed05d8 // indirect
 	github.com/smartcontractkit/chainlink-solana v1.3.1-0.20260605202330-b5a89c32fdc1 // indirect
 	github.com/smartcontractkit/chainlink-sui/codec v0.0.0-20260714120433-7667cad5ff5c // indirect
 	github.com/smartcontractkit/chainlink-testing-framework/framework v0.16.6-0.20260708113039-95f97b2d25e9 // indirect

integration-tests/load/go.mod

@@ -509,7 +509,7 @@ require (
 	github.com/smartcontractkit/chainlink-protos/ring/go v0.0.0-20260331131315-f08a616d8dcd // indirect
 	github.com/smartcontractkit/chainlink-protos/storage-service v0.3.0 // indirect
 	github.com/smartcontractkit/chainlink-protos/svr v1.3.0 // indirect
-	github.com/smartcontractkit/chainlink-protos/workflows/go v0.0.0-20260630073003-fb8da7229930 // indirect
+	github.com/smartcontractkit/chainlink-protos/workflows/go v0.0.0-20260716141634-c0cc05ed05d8 // indirect
 	github.com/smartcontractkit/chainlink-solana v1.3.1-0.20260605202330-b5a89c32fdc1 // indirect
 	github.com/smartcontractkit/chainlink-sui v0.0.0-20260713221039-69796c8a78ae // indirect
 	github.com/smartcontractkit/chainlink-sui/codec v0.0.0-20260714120433-7667cad5ff5c // indirect