[CRE] [5/5] Wire confidential workflow execution into CRE

Wires PRs 1-4 together and adds system test support:

- cre.go: start confidential relay service when config is enabled
- models.go: Attributes field on WorkflowSpec
- System tests: ConfidentialRelay feature plugin, workflow registration
  with attributes, CompileAndDeployConfidentialWorkflow helper

This PR includes code from PRs 1-4 for compilation. As those merge,
rebasing this branch will shrink the diff to just the wiring.

Depends on all of: #21635 PRs 1-4
Part of #21635

Author: nadahalli

.changeset/confidential-workflow-execution.md

@@ -0,0 +1,5 @@
+---
+"chainlink": minor
+---
+
+Add confidential workflow execution: ConfidentialModule, relay handler, gateway wiring, single-DON capability fix #added #db_update

core/capabilities/launcher.go

@@ -46,7 +46,7 @@ var defaultStreamConfig = p2ptypes.StreamConfig{
 
 type launcher struct {
 	services.StateMachine
-	lggr                logger.SugaredLogger
+	lggr                logger.Logger
 	myPeerID            p2ptypes.PeerID
 	peerWrapper         p2ptypes.PeerWrapper
 	dispatcher          remotetypes.Dispatcher
@@ -106,7 +106,7 @@ func NewLauncher(
 		return nil, fmt.Errorf("failed to create launcher metrics: %w", err)
 	}
 	return &launcher{
-		lggr:        logger.Sugared(lggr).Named("CapabilitiesLauncher"),
+		lggr:        logger.Named(lggr, "CapabilitiesLauncher"),
 		peerWrapper: peerWrapper,
 		dispatcher:  dispatcher,
 		cachedShims: cachedShims{
@@ -342,12 +342,9 @@ func (w *launcher) OnNewRegistry(ctx context.Context, localRegistry *registrysyn
 	for family := range myDONFamiliesSet {
 		myDONFamilies = append(myDONFamilies, family)
 	}
-	w.lggr.Debugw("Found my DON families", "count", len(myDONFamilies))
-	w.lggr.Tracew("My DON families", "myDONFamilies", myDONFamilies)
-	w.lggr.Debugw("Found my workflow DONs", "count", len(myWorkflowDONs))
-	w.lggr.Tracew("My workflow DONs", "myWorkflowDONs", myWorkflowDONs)
-	w.lggr.Debugw("Found all remote workflow DONs", "count", len(remoteWorkflowDONs))
-	w.lggr.Tracew("All remote workflow DONs", "remoteWorkflowDONs", remoteWorkflowDONs)
+	w.lggr.Debugw("Found my DON families", "count", len(myDONFamilies), "myDONFamilies", myDONFamilies)
+	w.lggr.Debugw("Found my workflow DONs", "count", len(myWorkflowDONs), "myWorkflowDONs", myWorkflowDONs)
+	w.lggr.Debugw("Found all remote workflow DONs", "count", len(remoteWorkflowDONs), "remoteWorkflowDONs", remoteWorkflowDONs)
 
 	// Capability DONs (with IsPublic = true) the current node is a part of.
 	// These need server-side shims to expose my own capabilities externally.
@@ -362,18 +359,14 @@ func (w *launcher) OnNewRegistry(ctx context.Context, localRegistry *registrysyn
 			}
 		}
 	}
-	w.lggr.Debugw("Found my capability DONs", "count", len(myCapabilityDONs))
-	w.lggr.Tracew("My capability DONs", "myCapabilityDONs", myCapabilityDONs)
-	w.lggr.Debugw("Found all remote capability DONs", "count", len(remoteCapabilityDONs))
-	w.lggr.Tracew("All remote capability DONs", "remoteCapabilityDONs", remoteCapabilityDONs)
+	w.lggr.Debugw("Found my capability DONs", "count", len(myCapabilityDONs), "myCapabilityDONs", myCapabilityDONs)
+	w.lggr.Debugw("Found all remote capability DONs", "count", len(remoteCapabilityDONs), "remoteCapabilityDONs", remoteCapabilityDONs)
 
 	if len(myDONFamilies) > 0 {
 		remoteWorkflowDONs = filterDONsByFamilies(remoteWorkflowDONs, myDONFamilies)
 		remoteCapabilityDONs = filterDONsByFamilies(remoteCapabilityDONs, myDONFamilies)
-		w.lggr.Debugw("Filtered remote workflow DONs to my families", "count", len(remoteWorkflowDONs))
-		w.lggr.Tracew("Filtered remote workflow DONs to my families", "remoteWorkflowDONs", remoteWorkflowDONs)
-		w.lggr.Debugw("Filtered remote capability DONs to my families", "count", len(remoteCapabilityDONs))
-		w.lggr.Tracew("Filtered remote capability DONs to my families", "remoteCapabilityDONs", remoteCapabilityDONs)
+		w.lggr.Debugw("Filtered remote workflow DONs to my families", "count", len(remoteWorkflowDONs), "remoteWorkflowDONs", remoteWorkflowDONs)
+		w.lggr.Debugw("Filtered remote capability DONs to my families", "count", len(remoteCapabilityDONs), "remoteCapabilityDONs", remoteCapabilityDONs)
 	} else {
 		// legacy / Keystone setting
 		w.lggr.Debug("My node doesn't belong to any DON families. No filtering will be applied.")
@@ -408,8 +401,23 @@ func (w *launcher) OnNewRegistry(ctx context.Context, localRegistry *registrysyn
 
 	belongsToACapabilityDON := len(myCapabilityDONs) > 0
 	if belongsToACapabilityDON {
+		// Include both remote workflow DONs and the node's own workflow DONs.
+		// In single-DON topologies (e.g. local CRE), the same DON is both a
+		// workflow DON and a capability DON, so remoteWorkflowDONs is empty.
+		// Without including myWorkflowDONs, capabilities fail to serve with
+		// "empty workflowDONs provided".
+		allWorkflowDONs := make([]registrysyncer.DON, 0, len(remoteWorkflowDONs)+len(myWorkflowDONs))
+		allWorkflowDONs = append(allWorkflowDONs, remoteWorkflowDONs...)
+		allWorkflowDONs = append(allWorkflowDONs, myWorkflowDONs...)
 		for _, myDON := range myCapabilityDONs {
-			w.serveCapabilities(ctx, w.myPeerID, myDON, localRegistry, remoteWorkflowDONs)
+			w.serveCapabilities(ctx, w.myPeerID, myDON, localRegistry, allWorkflowDONs)
+
+			// Capability DONs also need remote capabilities (e.g. relay DON
+			// needs vault for secret fetching). Without this, only workflow
+			// DONs discover cross-DON capabilities.
+			for _, rcd := range remoteCapabilityDONs {
+				w.addRemoteCapabilities(ctx, myDON, rcd, localRegistry)
+			}
 		}
 	}
 
@@ -425,7 +433,7 @@ func (w *launcher) OnNewRegistry(ctx context.Context, localRegistry *registrysyn
 	}
 	if w.don2donSharedPeer != nil {
 		donPairs := w.donPairsToUpdate(w.myPeerID, localRegistry)
-		err := w.don2donSharedPeer.UpdateConnectionsByDONs(ctx, donPairs, w.p2pStreamConfig)
+		err := w.don2donSharedPeer.UpdateConnectionsByDONs(ctx, donPairs, defaultStreamConfig)
 		if err != nil {
 			return fmt.Errorf("failed to update peer connections: %w", err)
 		}

core/capabilities/launcher_test.go

@@ -12,7 +12,6 @@ import (
 	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/types/known/durationpb"
 
-	"github.com/smartcontractkit/libocr/ragep2p"
 	ragetypes "github.com/smartcontractkit/libocr/ragep2p/types"
 
 	"github.com/smartcontractkit/chainlink-common/pkg/capabilities"
@@ -234,6 +233,73 @@ func TestLauncher(t *testing.T) {
 		assert.Equal(t, 1, observedLogs.FilterMessage("failed to serve capability").Len())
 	})
 
+	t.Run("OK-single_DON_serves_capabilities", func(t *testing.T) {
+		// Regression test: in single-DON topologies (e.g. local CRE), the same
+		// DON is both a workflow DON and a capability DON. The DON appears in
+		// myWorkflowDONs (not remoteWorkflowDONs). Previously, serveCapabilities
+		// only received remoteWorkflowDONs, causing executable.Server.SetConfig
+		// to fail with "empty workflowDONs provided".
+		lggr, observedLogs := logger.TestObserved(t, zapcore.DebugLevel)
+		registry := NewRegistry(lggr)
+		dispatcher := remoteMocks.NewDispatcher(t)
+
+		nodes := newNodes(4)
+		peer := mocks.NewPeer(t)
+		peer.On("UpdateConnections", mock.Anything).Return(nil)
+		peer.On("ID").Return(nodes[0])
+		peer.On("IsBootstrap").Return(false)
+		wrapper := mocks.NewPeerWrapper(t)
+		wrapper.On("GetPeer").Return(peer)
+
+		fullTriggerCapID := "streams-trigger@1.0.0"
+		mt := newMockTrigger(capabilities.MustNewCapabilityInfo(
+			fullTriggerCapID,
+			capabilities.CapabilityTypeTrigger,
+			"streams trigger",
+		))
+		require.NoError(t, registry.Add(t.Context(), mt))
+
+		fullTargetID := "write-chain_evm_1@1.0.0"
+		mtarg := &mockCapability{
+			CapabilityInfo: capabilities.MustNewCapabilityInfo(
+				fullTargetID,
+				capabilities.CapabilityTypeTarget,
+				"write chain",
+			),
+		}
+		require.NoError(t, registry.Add(t.Context(), mtarg))
+
+		triggerCapID := RandomUTF8BytesWord()
+		targetCapID := RandomUTF8BytesWord()
+
+		// Single DON: acceptsWorkflows=true AND has capability configurations.
+		// This puts it in both myWorkflowDONs and myCapabilityDONs.
+		dID := uint32(1)
+		localRegistry := buildLocalRegistry()
+		addDON(localRegistry, dID, uint32(0), uint8(1), true, true, nodes, []string{"zone-a"}, 1, [][32]byte{triggerCapID, targetCapID})
+		addCapabilityToDON(localRegistry, dID, fullTriggerCapID, capabilities.CapabilityTypeTrigger, nil)
+		addCapabilityToDON(localRegistry, dID, fullTargetID, capabilities.CapabilityTypeTarget, nil)
+
+		launcher, err := NewLauncher(
+			lggr,
+			wrapper,
+			nil,
+			nil,
+			dispatcher,
+			registry,
+			&mockDonNotifier{},
+		)
+		require.NoError(t, err)
+		require.NoError(t, launcher.Start(t.Context()))
+		defer launcher.Close()
+
+		dispatcher.On("SetReceiver", fullTriggerCapID, dID, mock.AnythingOfType("*remote.triggerPublisher")).Return(nil)
+		dispatcher.On("SetReceiver", fullTargetID, dID, mock.AnythingOfType("*executable.server")).Return(nil)
+
+		require.NoError(t, launcher.OnNewRegistry(t.Context(), localRegistry))
+		assert.Equal(t, 0, observedLogs.FilterMessage("failed to serve capability").Len())
+	})
+
 	t.Run("start and close with nil peer wrapper", func(t *testing.T) {
 		lggr := logger.Test(t)
 		registry := NewRegistry(lggr)
@@ -893,24 +959,10 @@ func TestLauncher_V2CapabilitiesAddViaCombinedClient(t *testing.T) {
 	addCapabilityToDON(localRegistry, zoneBDonID, fullExecutableCapID, capabilities.CapabilityTypeTarget, execCfg)
 	addCapabilityToDON(localRegistry, capDonID, fullLocalCapID, capabilities.CapabilityTypeAction, localCfg) // should be skipped
 
-	customStreamConfig := p2ptypes.StreamConfig{
-		IncomingMessageBufferSize: 999,
-		OutgoingMessageBufferSize: 888,
-		MaxMessageLenBytes:        777777,
-		MessageRateLimiter: ragep2p.TokenBucketParams{
-			Rate:     50.0,
-			Capacity: 250,
-		},
-		BytesRateLimiter: ragep2p.TokenBucketParams{
-			Rate:     2500000.0,
-			Capacity: 5000000,
-		},
-	}
-
 	sharedPeer := mocks.NewSharedPeer(t)
 	sharedPeer.On("ID").Return(workflowDonNodes[0])
 	sharedPeer.On("IsBootstrap").Return(false)
-	sharedPeer.On("UpdateConnectionsByDONs", mock.Anything, mock.Anything, customStreamConfig).Return(nil)
+	sharedPeer.On("UpdateConnectionsByDONs", mock.Anything, mock.Anything, mock.Anything).Return(nil)
 
 	launcher, err := NewLauncher(
 		lggr,
@@ -922,7 +974,6 @@ func TestLauncher_V2CapabilitiesAddViaCombinedClient(t *testing.T) {
 		&mockDonNotifier{},
 	)
 	require.NoError(t, err)
-	launcher.p2pStreamConfig = customStreamConfig
 	servicetest.Run(t, launcher)
 
 	dispatcher.On("SetReceiverForMethod", fullTriggerCapID, capDonID, "StreamsTrigger", mock.AnythingOfType("*remote.triggerSubscriber")).Return(nil)

core/scripts/cre/environment/environment/workflow.go

@@ -475,7 +475,7 @@ func deployWorkflow(
 
 	fmt.Printf("\n⚙️ Registering workflow '%s' with the workflow registry\n\n", workflowNameFlag)
 
-	workflowID, registerErr := creworkflow.RegisterWithContract(ctx, sethClient, common.HexToAddress(workflowRegistryAddress), workflowRegistryVersion, uint64(donIDFlag), workflowNameFlag, "file://"+wasmWorkflowFilePathFlag, configPath, secretsPath, &containerTargetDirFlag)
+	workflowID, registerErr := creworkflow.RegisterWithContract(ctx, sethClient, common.HexToAddress(workflowRegistryAddress), workflowRegistryVersion, uint64(donIDFlag), workflowNameFlag, "file://"+wasmWorkflowFilePathFlag, configPath, secretsPath, nil, &containerTargetDirFlag)
 	if registerErr != nil {
 		return errors.Wrapf(registerErr, "❌ failed to register workflow %s", workflowNameFlag)
 	}

core/scripts/go.mod

@@ -46,13 +46,13 @@ require (
 	github.com/shopspring/decimal v1.4.0
 	github.com/smartcontractkit/chainlink-automation v0.8.1
 	github.com/smartcontractkit/chainlink-ccip v0.1.1-solana.0.20260317185256-d5f7db87ae70
-	github.com/smartcontractkit/chainlink-common v0.11.0
+	github.com/smartcontractkit/chainlink-common v0.11.1-0.20260323163826-2c5b95089478
 	github.com/smartcontractkit/chainlink-common/keystore v1.0.2
 	github.com/smartcontractkit/chainlink-data-streams v0.1.13
 	github.com/smartcontractkit/chainlink-deployments-framework v0.86.3
 	github.com/smartcontractkit/chainlink-evm v0.3.4-0.20260320152158-2191d797b5ce
 	github.com/smartcontractkit/chainlink-evm/gethwrappers v0.0.0-20260119171452-39c98c3b33cd
-	github.com/smartcontractkit/chainlink-protos/cre/go v0.0.0-20260226130359-963f935e0396
+	github.com/smartcontractkit/chainlink-protos/cre/go v0.0.0-20260320153346-314ec8dbe5a4
 	github.com/smartcontractkit/chainlink-protos/job-distributor v0.18.0
 	github.com/smartcontractkit/chainlink-testing-framework/framework v0.15.5
 	github.com/smartcontractkit/chainlink-testing-framework/framework/components/dockercompose v0.1.20
@@ -345,6 +345,7 @@ require (
 	github.com/hashicorp/yamux v0.1.2 // indirect
 	github.com/hasura/go-graphql-client v0.15.1 // indirect
 	github.com/hdevalence/ed25519consensus v0.2.0 // indirect
+	github.com/hf/nitrite v0.0.0-20241225144000-c2d5d3c4f303 // indirect
 	github.com/holiman/billy v0.0.0-20250707135307-f2f9b9aae7db // indirect
 	github.com/holiman/bloomfilter/v2 v2.0.3 // indirect
 	github.com/holiman/uint256 v1.3.2 // indirect
@@ -493,6 +494,7 @@ require (
 	github.com/smartcontractkit/chainlink-ccip/deployment v0.0.0-20260317185256-d5f7db87ae70 // indirect
 	github.com/smartcontractkit/chainlink-ccv v0.0.0-20260320145055-eb20b529ff95 // indirect
 	github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20251211140724-319861e514c4 // indirect
+	github.com/smartcontractkit/chainlink-common/pkg/teeattestation v0.0.0-20260316172927-2c727f64397c // indirect
 	github.com/smartcontractkit/chainlink-evm/contracts/cre/gobindings v0.0.0-20260107191744-4b93f62cffe3 // indirect
 	github.com/smartcontractkit/chainlink-feeds v0.1.2-0.20250227211209-7cd000095135 // indirect
 	github.com/smartcontractkit/chainlink-framework/capabilities v0.0.0-20250818175541-3389ac08a563 // indirect

core/scripts/go.sum

@@ -32,6 +32,7 @@ import (
 
 	"github.com/smartcontractkit/chainlink/v2/core/capabilities"
 	"github.com/smartcontractkit/chainlink/v2/core/capabilities/compute"
+	"github.com/smartcontractkit/chainlink/v2/core/capabilities/confidentialrelay"
 	gatewayconnector "github.com/smartcontractkit/chainlink/v2/core/capabilities/gateway_connector"
 	"github.com/smartcontractkit/chainlink/v2/core/capabilities/localcapmgr"
 	"github.com/smartcontractkit/chainlink/v2/core/capabilities/remote"
@@ -169,6 +170,17 @@ func (s *Services) newSubservices(
 		}
 		s.GatewayConnectorWrapper = gatewayConnectorWrapper
 		srvs = append(srvs, gatewayConnectorWrapper)
+
+		if relayConfig := cfg.CRE().ConfidentialRelay(); relayConfig.Enabled() {
+			relayService := confidentialrelay.NewService(
+				gatewayConnectorWrapper,
+				opts.CapabilitiesRegistry,
+				[]byte(relayConfig.TrustedPCRs()),
+				relayConfig.CARootsPEM(),
+				lggr,
+			)
+			srvs = append(srvs, relayService)
+		}
 	}
 
 	if cfg.CRE().Linking().URL() != "" {