[pull] develop from smartcontractkit:develop

core/scripts/cre/environment/configs/mixed-env-confidential-workflows.toml.tmpl

@@ -246,12 +246,11 @@
     image = "postgres:15.0"
     port = 13200
 
-  # The bootstrap/gateway node does not run the reporting plugin, so its code
-  # version does not affect the non-determinism signal. Keep it on develop for stability.
+  # The Gateway DON uses only 1 node per request, so it should run the PR image, exercise the code in PR.
   [[nodesets.node_specs]]
     roles = ["bootstrap", "gateway"]
     [nodesets.node_specs.node]
-      image = "${CRE_BASELINE_IMAGE}"
+      image = "${CRE_PR_IMAGE}"
       pull_image = ${CRE_PULL_IMAGE}
       # 5002 is the web API capabilities port for incoming requests
       # 15002 is the vault port for incoming requests

core/scripts/go.mod

@@ -33,7 +33,7 @@ require (
 	github.com/shopspring/decimal v1.4.0
 	github.com/smartcontractkit/chain-selectors v1.0.111
 	github.com/smartcontractkit/chainlink-ccip/chains/evm v0.0.0-20260916222901-720a003dab50
-	github.com/smartcontractkit/chainlink-common v0.11.2-0.20260921174556-538f4bf0cb6f
+	github.com/smartcontractkit/chainlink-common v0.11.2-0.20260928161532-9b1062f5076f
 	github.com/smartcontractkit/chainlink-common/keystore v1.3.1-0.20260903141829-ef07b52a737d
 	github.com/smartcontractkit/chainlink-deployments-framework v0.122.2
 	github.com/smartcontractkit/chainlink-evm v0.3.4-0.20260917132551-2a2b38e8b744
@@ -210,7 +210,6 @@ require (
 	github.com/docker/docker-credential-helpers v0.9.8 // indirect
 	github.com/docker/go-connections v0.7.0 // indirect
 	github.com/docker/go-units v0.5.0 // indirect
-	github.com/dominikbraun/graph v0.23.0 // indirect
 	github.com/doyensec/safeurl v0.2.5 // indirect
 	github.com/dustin/go-humanize v1.0.1 // indirect
 	github.com/dvsekhvalnov/jose2go v1.7.0 // indirect

core/services/workflows/cmd/cre/utils/standalone_engine.go

@@ -176,7 +176,7 @@ func NewStandaloneEngine(
 		Request:         &sdkpb.ExecuteRequest_Subscribe{},
 		MaxResponseSize: uint64(moduleExecuteMaxResponseSizeBytes),
 		Config:          config,
-	}, v2.NewDisallowedExecutionHelper(lggr, nil, &types.LocalTimeProvider{}, secretsFetcher))
+	}, v2.NewDisallowedExecutionHelper(lggr, nil, &types.LocalTimeProvider{}))
 	if err != nil {
 		return nil, nil, fmt.Errorf("failed to execute subscribe: %w", err)
 	}

core/services/workflows/monitoring/monitoring.go

@@ -31,6 +31,7 @@ type EngineMetrics struct {
 	workflowExecutionLatencyGauge            metric.Int64Gauge // ms
 	workflowStepErrorCounter                 metric.Int64Counter
 	workflowInitializationCounter            metric.Int64Counter
+	workflowInitializationFailureCounter     metric.Int64Counter
 	workflowTriggerEventErrorCounter         metric.Int64Counter
 	workflowTriggerEventQueueFullCounter     metric.Int64Counter
 
@@ -164,6 +165,13 @@ func InitMonitoringResources() (em *EngineMetrics, err error) {
 		return nil, fmt.Errorf("failed to register workflow initialization counter: %w", err)
 	}
 
+	em.workflowInitializationFailureCounter, err = beholder.GetMeter().Int64Counter(
+		"platform_engine_workflow_initialization_failures_total",
+		metric.WithDescription("Count of failed engine initializations by failure reason"))
+	if err != nil {
+		return nil, fmt.Errorf("failed to register workflow initialization failure counter: %w", err)
+	}
+
 	em.workflowStepErrorCounter, err = beholder.GetMeter().Int64Counter("platform_engine_workflow_errors")
 	if err != nil {
 		return nil, fmt.Errorf("failed to register workflow step error counter: %w", err)
@@ -633,6 +641,18 @@ func (c WorkflowsMetricLabeler) IncrementWorkflowInitializationCounter(ctx conte
 	c.em.workflowInitializationCounter.Add(ctx, 1, metric.WithAttributes(otelLabels...))
 }
 
+// IncrementWorkflowInitializationFailureCounter records one failed engine
+// initialization. reason must be a low-cardinality value identifying the
+// failing phase; see the initFailure* constants in the engine package.
+func (c WorkflowsMetricLabeler) IncrementWorkflowInitializationFailureCounter(ctx context.Context, reason string) {
+	if reason == "" {
+		reason = "unknown"
+	}
+	otelLabels := beholder.OtelAttributes(c.Labels).AsStringAttributes()
+	otelLabels = append(otelLabels, attribute.String("reason", reason))
+	c.em.workflowInitializationFailureCounter.Add(ctx, 1, metric.WithAttributes(otelLabels...))
+}
+
 func (c WorkflowsMetricLabeler) IncrementWorkflowTriggerEventErrorCounter(ctx context.Context) {
 	otelLabels := beholder.OtelAttributes(c.Labels).AsStringAttributes()
 	c.em.workflowTriggerEventErrorCounter.Add(ctx, 1, metric.WithAttributes(otelLabels...))

core/services/workflows/shardownership/shard_index_mapper.go

@@ -11,13 +11,9 @@ import (
 	commoncap "github.com/smartcontractkit/chainlink-common/pkg/capabilities"
 	"github.com/smartcontractkit/chainlink-common/pkg/capabilities/registry"
 	"github.com/smartcontractkit/chainlink-common/pkg/logger"
+	"github.com/smartcontractkit/chainlink/v2/core/services/workflows/shardownership/types"
 )
 
-// shardNameMarker is the marker a workflow DON's name is suffixed with to
-// encode its shard index (e.g. "workflow-1-zone-a-shard-1" -> shard 1). A
-// name with no such suffix (e.g. "workflow-1-zone-a") is shard index 0.
-const shardNameMarker = "shard-"
-
 // ShardIndexMapper tracks the workflow DONs currently known to the Capabilities
 // Registry and detects their shard indices (0, 1, 2, ...).
 // It's a global singleton object used by ManualShardResolver and, in consequence,
@@ -117,13 +113,13 @@ func shardDONsByIndex(reg *registry.RegistryMetadata, localDON commoncap.DON, na
 // workflow DON's name (e.g. "workflow-1-zone-a-shard-1" -> 1). A name with no
 // "shard-" suffix is shard index 0.
 func shardIndexFromName(name string) (uint32, error) {
-	idx := strings.LastIndex(name, shardNameMarker)
+	idx := strings.LastIndex(name, types.ShardNameMarker)
 	if idx == -1 {
 		return 0, nil
 	}
-	suffix := name[idx+len(shardNameMarker):]
+	suffix := name[idx+len(types.ShardNameMarker):]
 	if len(suffix) != 1 || suffix[0] < '0' || suffix[0] > '9' {
-		return 0, fmt.Errorf("expected DON name %q to end with %q followed by a single digit, got suffix %q", name, shardNameMarker, suffix)
+		return 0, fmt.Errorf("expected DON name %q to end with %q followed by a single digit, got suffix %q", name, types.ShardNameMarker, suffix)
 	}
 	return uint32(suffix[0] - '0'), nil
 }
@@ -134,7 +130,7 @@ func shardIndexFromName(name string) (uint32, error) {
 // "workflow-1-zone-a-shard-1" -> "workflow-1-zone-a"). A name with no such
 // suffix is returned unchanged.
 func shardGroupNamePrefix(name string) string {
-	idx := strings.LastIndex(name, shardNameMarker)
+	idx := strings.LastIndex(name, types.ShardNameMarker)
 	if idx == -1 {
 		return name
 	}

core/services/workflows/shardownership/types/types.go

@@ -0,0 +1,15 @@
+package types
+
+const (
+	// ShardNameMarker is the marker a workflow DON's name is suffixed with to
+	// encode its shard index (e.g. "workflow-1-zone-a-shard-1" -> shard 1). A
+	// name with no such suffix (e.g. "workflow-1-zone-a") is shard index 0.
+	ShardNameMarker = "shard-"
+
+	// WorkflowFamilySuffix is the suffix appended to a DON family name to
+	// denote its dedicated workflow-fetching family. A DON belonging to a
+	// family named "<name>WorkflowFamilySuffix" (e.g. "zone-a_workflows")
+	// should also fetch workflows registered under the base family with the
+	// suffix stripped (e.g. "zone-a").
+	WorkflowFamilySuffix = "_workflows"
+)

core/services/workflows/syncer/v2/activation_errors.go

@@ -7,6 +7,7 @@ import (
 
 	eventsv2 "github.com/smartcontractkit/chainlink-protos/workflows/go/v2"
 	"github.com/smartcontractkit/chainlink/v2/core/services/workflows/types"
+	wfv2 "github.com/smartcontractkit/chainlink/v2/core/services/workflows/v2"
 )
 
 // ActivationRetryPolicy controls how failed workflow activations are retried.
@@ -96,7 +97,12 @@ func isPermanentEngineInitError(err error) bool {
 			// cron failure when it can't parse the schedule
 			strings.Contains(msg, "failed to initialize job"),
 			// cron rejection when the schedule fires faster than the allowed minimum
-			strings.Contains(msg, "maximum fastest cron schedule"):
+			strings.Contains(msg, "maximum fastest cron schedule"),
+			// guest attempted a disallowed host call (capability or secrets) during
+			// the trigger subscription phase; a deterministic user error that can
+			// never succeed on retry
+			strings.Contains(msg, wfv2.ErrCapabilityCallDuringSubscription.Error()),
+			strings.Contains(msg, wfv2.ErrSecretsCallDuringSubscription.Error()):
 			return true
 		}
 		err = errors.Unwrap(err)

core/services/workflows/syncer/v2/activation_errors_test.go

@@ -8,6 +8,7 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/smartcontractkit/chainlink/v2/core/services/workflows/types"
+	wfv2 "github.com/smartcontractkit/chainlink/v2/core/services/workflows/v2"
 )
 
 func Test_classifyActivationError(t *testing.T) {
@@ -60,6 +61,20 @@ func Test_classifyActivationError(t *testing.T) {
 				errors.New("[3]InvalidArgument: maximum fastest cron schedule is 30s")),
 			want: ActivationNonRetryable,
 		},
+		{
+			// mirrors the engine's wrapping of the guest error string: the
+			// host rejects the secrets call, the guest reports it back and
+			// Subscribe fails
+			name: "secrets call during trigger subscription is non-retryable",
+			err: fmt.Errorf("failed to execute subscribe: failed to get secrets for call 1: %s",
+				wfv2.ErrSecretsCallDuringSubscription.Error()),
+			want: ActivationNonRetryable,
+		},
+		{
+			name: "capability call during trigger subscription is non-retryable",
+			err:  fmt.Errorf("failed to execute subscribe: %s", wfv2.ErrCapabilityCallDuringSubscription.Error()),
+			want: ActivationNonRetryable,
+		},
 		{
 			name: "unknown error is retryable",
 			err:  errors.New("unexpected engine failure"),

core/services/workflows/syncer/v2/contract_workflow_source.go

@@ -72,7 +72,7 @@ func (c *ContractWorkflowSource) ListWorkflowMetadata(ctx context.Context, don c
 	var headAtLastRead *commontypes.Head
 	var allWorkflows []WorkflowMetadataView
 
-	for _, family := range don.Families {
+	for _, family := range ExpandWorkflowFamilies(don.Families) {
 		params := GetWorkflowListByDONParams{
 			DonFamily: family,
 			Start:     big.NewInt(0),

core/services/workflows/syncer/v2/evictable_module.go

@@ -220,10 +220,6 @@ func (m *EvictableModule) Close() {
 	}
 }
 
-func (m *EvictableModule) IsLegacyDAG() bool {
-	return false
-}
-
 func (m *EvictableModule) Execute(ctx context.Context, request *sdkpb.ExecuteRequest, handler host.ExecutionHelper) (*sdkpb.ExecutionResult, error) {
 	// Each loaded module is held behind a refcounted loadedModule. Pinning is
 	// a CAS-conditional refcount increment: it succeeds only if the holder is

core/services/workflows/syncer/v2/evictable_module_test.go

@@ -44,9 +44,8 @@ type fakeModule struct {
 	closeCalls atomic.Int32
 }
 
-func (f *fakeModule) Start()            {}
-func (f *fakeModule) IsLegacyDAG() bool { return false }
-func (f *fakeModule) Close()            { f.closeCalls.Add(1) }
+func (f *fakeModule) Start() {}
+func (f *fakeModule) Close() { f.closeCalls.Add(1) }
 func (f *fakeModule) Execute(_ context.Context, _ *sdkpb.ExecuteRequest, _ host.ExecutionHelper) (*sdkpb.ExecutionResult, error) {
 	return &sdkpb.ExecutionResult{}, nil
 }
@@ -180,7 +179,6 @@ func TestEvictable_DelegatesToInner(t *testing.T) { //nolint:paralleltest // pac
 	result, err := em.Execute(context.Background(), &sdkpb.ExecuteRequest{}, nil)
 	require.NoError(t, err)
 	assert.NotNil(t, result)
-	assert.False(t, em.IsLegacyDAG())
 }
 
 func TestEvictable_LastUsedUpdated(t *testing.T) { //nolint:paralleltest // package-level hooks and eviction config

core/services/workflows/syncer/v2/file_workflow_source.go

@@ -107,7 +107,7 @@ func (f *FileWorkflowSource) ListWorkflowMetadata(ctx context.Context, don capab
 
 	// Build a set of DON families for efficient lookup
 	donFamilySet := make(map[string]bool)
-	for _, family := range don.Families {
+	for _, family := range ExpandWorkflowFamilies(don.Families) {
 		donFamilySet[family] = true
 	}
 

core/services/workflows/syncer/v2/grpc_workflow_source.go

@@ -173,10 +173,11 @@ func (g *GRPCWorkflowSource) ListWorkflowMetadata(ctx context.Context, don capab
 
 	var allViews []WorkflowMetadataView
 	var start int64
+	families := ExpandWorkflowFamilies(don.Families)
 
 	// Fetch all pages
 	for {
-		workflows, hasMore, err := g.fetchPageWithRetry(ctx, don.Families, start)
+		workflows, hasMore, err := g.fetchPageWithRetry(ctx, families, start)
 		if err != nil {
 			return nil, nil, err
 		}

core/services/workflows/syncer/v2/handler.go

@@ -422,7 +422,7 @@ func (h *eventHandler) close() error {
 	// untouched. A spec that is genuinely released stops being snapshotted
 	// (the spec meter sub-service unregisters itself after this hook runs).
 	cs := make([]io.Closer, 0, len(es)+1)
-	cs = append(cs, h.engineLimiters)
+	cs = append(cs, h.engineLimiters, h.featureFlags)
 	for _, e := range es {
 		cs = append(cs, e)
 	}

core/services/workflows/syncer/v2/helpers.go

@@ -3,6 +3,7 @@ package v2
 import (
 	"context"
 	"errors"
+	"strings"
 	"sync"
 
 	"github.com/ethereum/go-ethereum/common/hexutil"
@@ -13,9 +14,38 @@ import (
 	eventsv2 "github.com/smartcontractkit/chainlink-protos/workflows/go/v2"
 	"github.com/smartcontractkit/chainlink/v2/core/services/job"
 	"github.com/smartcontractkit/chainlink/v2/core/services/workflows/ratelimiter"
+	shardownershiptypes "github.com/smartcontractkit/chainlink/v2/core/services/workflows/shardownership/types"
 	"github.com/smartcontractkit/chainlink/v2/core/services/workflows/syncerlimiter"
 )
 
+// ExpandWorkflowFamilies returns families with an extra entry added for every
+// family ending in shardownershiptypes.WorkflowFamilySuffix (e.g.
+// "zone-a_workflows"): the base family name with that suffix stripped (e.g.
+// "zone-a"). This lets a DON belonging to a "<name>_workflows" family also
+// fetch workflows registered directly under the bare "<name>" family. Order
+// is preserved and duplicates are dropped.
+func ExpandWorkflowFamilies(families []string) []string {
+	seen := make(map[string]bool, len(families))
+	expanded := make([]string, 0, len(families))
+
+	add := func(family string) {
+		if seen[family] {
+			return
+		}
+		seen[family] = true
+		expanded = append(expanded, family)
+	}
+
+	for _, family := range families {
+		add(family)
+		if base, ok := strings.CutSuffix(family, shardownershiptypes.WorkflowFamilySuffix); ok {
+			add(base)
+		}
+	}
+
+	return expanded
+}
+
 var rlConfig = ratelimiter.Config{
 	GlobalRPS:      1000.0,
 	GlobalBurst:    1000,

core/services/workflows/syncer/v2/helpers_test.go

@@ -0,0 +1,49 @@
+package v2
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestExpandWorkflowFamilies(t *testing.T) {
+	t.Parallel()
+	tests := []struct {
+		name     string
+		families []string
+		want     []string
+	}{
+		{
+			name:     "no workflows suffix",
+			families: []string{"zone-a"},
+			want:     []string{"zone-a"},
+		},
+		{
+			name:     "workflows suffix expands to base family",
+			families: []string{"zone-a_workflows"},
+			want:     []string{"zone-a_workflows", "zone-a"},
+		},
+		{
+			name:     "base family already present is not duplicated",
+			families: []string{"zone-a_workflows", "zone-a"},
+			want:     []string{"zone-a_workflows", "zone-a"},
+		},
+		{
+			name:     "mixed families",
+			families: []string{"zone-a", "zone-b_workflows"},
+			want:     []string{"zone-a", "zone-b_workflows", "zone-b"},
+		},
+		{
+			name:     "empty input",
+			families: []string{},
+			want:     []string{},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+			assert.Equal(t, tt.want, ExpandWorkflowFamilies(tt.families))
+		})
+	}
+}