From fa2bbd3a63b1286bcda96ca80086ec70659f8861 Mon Sep 17 00:00:00 2001 From: De Clercq Wentzel <10665586+wentzeld@users.noreply.github.com> Date: Wed, 21 Oct 2026 08:25:24 -0700 Subject: [PATCH 1/6] Emit vault get-secrets capability execution event for CRE UI Wrap ExecutionHelper.GetSecrets so each vault.secrets.get batch call emits a workflows.v2 CapabilityExecutionStarted/Finished pair, matching the events the engine already emits for other capability calls. The event uses the real workflow execution ID and finishes errored on failure. No new protos or consumer changes. --- .../workflows/v2/capability_executor.go | 28 ++++ .../v2/capability_executor_getsecrets_test.go | 125 ++++++++++++++++++ 2 files changed, 153 insertions(+) create mode 100644 core/services/workflows/v2/capability_executor_getsecrets_test.go diff --git a/core/services/workflows/v2/capability_executor.go b/core/services/workflows/v2/capability_executor.go index 965ecca08bd..7e0eaf6beb3 100644 --- a/core/services/workflows/v2/capability_executor.go +++ b/core/services/workflows/v2/capability_executor.go @@ -301,6 +301,34 @@ func (c *ExecutionHelper) callCapability(ctx context.Context, request *sdkpb.Cap }, nil } +// GetSecrets wraps the embedded SecretsFetcher so a Vault get-secrets call is +// observable in the CRE UI as a capability execution step, matching the events +// the engine already emits for other capability calls (see callCapability). +// One Started/Finished pair is emitted per batch call; a failed call finishes as +// errored with the returned error. +func (c *ExecutionHelper) GetSecrets(ctx context.Context, request *sdkpb.GetSecretsRequest) ([]*sdkpb.SecretResponse, error) { + stepRef := strconv.FormatInt(int64(request.GetCallbackId()), 10) + labels := c.base.eventLabels() + + _ = events.EmitCapabilityStartedEvent( + ctx, labels, c.WorkflowExecutionID, + vaultcommon.CapabilityID, stepRef, vaultcommon.MethodGetSecrets, + ) + + resp, err := c.SecretsFetcher.GetSecrets(ctx, request) + + status := store.StatusCompleted + if err != nil { + status = store.StatusErrored + } + _ = events.EmitCapabilityFinishedEvent( + ctx, labels, c.WorkflowExecutionID, + vaultcommon.CapabilityID, stepRef, status, vaultcommon.MethodGetSecrets, err, + ) + + return resp, err +} + func (c *ExecutionHelper) GetWorkflowExecutionID() string { return c.WorkflowExecutionID } diff --git a/core/services/workflows/v2/capability_executor_getsecrets_test.go b/core/services/workflows/v2/capability_executor_getsecrets_test.go new file mode 100644 index 00000000000..34ad1aeb9cd --- /dev/null +++ b/core/services/workflows/v2/capability_executor_getsecrets_test.go @@ -0,0 +1,125 @@ +package v2 + +import ( + "context" + "errors" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "google.golang.org/protobuf/proto" + + "github.com/smartcontractkit/chainlink-common/pkg/beholder" + "github.com/smartcontractkit/chainlink-common/pkg/beholder/beholdertest" + vaultcommon "github.com/smartcontractkit/chainlink-common/pkg/capabilities/actions/vault" + sdkpb "github.com/smartcontractkit/chainlink-protos/cre/go/sdk" + eventsv2 "github.com/smartcontractkit/chainlink-protos/workflows/go/v2" + "github.com/smartcontractkit/chainlink/v2/core/platform" +) + +const ( + entityCapabilityExecutionStartedV2 = "workflows.v2.CapabilityExecutionStarted" + entityCapabilityExecutionFinishedV2 = "workflows.v2.CapabilityExecutionFinished" +) + +// stubSecretsFetcher is a minimal SecretsFetcher for exercising ExecutionHelper.GetSecrets. +type stubSecretsFetcher struct { + resp []*sdkpb.SecretResponse + err error +} + +func (s stubSecretsFetcher) GetSecrets(context.Context, *sdkpb.GetSecretsRequest) ([]*sdkpb.SecretResponse, error) { + return s.resp, s.err +} + +// newGetSecretsTestHelper builds an ExecutionHelper whose base engine only carries +// the workflow event labels, which is all GetSecrets reads from it. +func newGetSecretsTestHelper(t *testing.T, executionID string, fetcher SecretsFetcher) *ExecutionHelper { + t.Helper() + + labels := map[string]string{ + platform.KeyWorkflowOwner: "0x0000000000000000000000000000000000000001", + platform.KeyWorkflowID: "wf-id", + platform.KeyWorkflowName: "wf-name", + } + be := &baseEngine{} + be.loggerLabels.Store(&labels) + + return &ExecutionHelper{ + base: be, + WorkflowExecutionID: executionID, + SecretsFetcher: fetcher, + } +} + +func decodeCapabilityExecutionStarted(t *testing.T, observer beholdertest.Observer) *eventsv2.CapabilityExecutionStarted { + t.Helper() + msgs := observer.Messages(t, beholder.AttrKeyEntity, entityCapabilityExecutionStartedV2) + require.Len(t, msgs, 1, "expected exactly one CapabilityExecutionStarted event") + var out eventsv2.CapabilityExecutionStarted + require.NoError(t, proto.Unmarshal(msgs[0].Body, &out)) + return &out +} + +func decodeCapabilityExecutionFinished(t *testing.T, observer beholdertest.Observer) *eventsv2.CapabilityExecutionFinished { + t.Helper() + msgs := observer.Messages(t, beholder.AttrKeyEntity, entityCapabilityExecutionFinishedV2) + require.Len(t, msgs, 1, "expected exactly one CapabilityExecutionFinished event") + var out eventsv2.CapabilityExecutionFinished + require.NoError(t, proto.Unmarshal(msgs[0].Body, &out)) + return &out +} + +// TestExecutionHelper_GetSecretsEmitsCapabilityExecutionEvents verifies that a +// successful vault get-secrets call emits a Started/Finished pair keyed to the +// real workflow execution ID, so it surfaces in the CRE UI like any other +// capability call. +// +// NewObserver mutates process-global beholder state, so this test must not be parallel. +func TestExecutionHelper_GetSecretsEmitsCapabilityExecutionEvents(t *testing.T) { + observer := beholdertest.NewObserver(t) + + const executionID = "exec-1" + callbackID := int32(7) + h := newGetSecretsTestHelper(t, executionID, stubSecretsFetcher{}) + + _, err := h.GetSecrets(t.Context(), &sdkpb.GetSecretsRequest{CallbackId: callbackID}) + require.NoError(t, err) + + started := decodeCapabilityExecutionStarted(t, observer) + assert.Equal(t, vaultcommon.CapabilityID, started.GetCapabilityID()) + assert.Equal(t, vaultcommon.MethodGetSecrets, started.GetMethod()) + assert.Equal(t, callbackID, started.GetStepRef()) + assert.Equal(t, executionID, started.GetWorkflowExecutionID()) + require.NotNil(t, started.GetWorkflow()) + assert.Equal(t, "wf-id", started.GetWorkflow().GetWorkflowID()) + assert.Equal(t, "wf-name", started.GetWorkflow().GetWorkflowName()) + assert.Equal(t, "0x0000000000000000000000000000000000000001", started.GetWorkflow().GetWorkflowOwner()) + + finished := decodeCapabilityExecutionFinished(t, observer) + assert.Equal(t, vaultcommon.CapabilityID, finished.GetCapabilityID()) + assert.Equal(t, vaultcommon.MethodGetSecrets, finished.GetMethod()) + assert.Equal(t, callbackID, finished.GetStepRef()) + assert.Equal(t, executionID, finished.GetWorkflowExecutionID()) + assert.Equal(t, eventsv2.ExecutionStatus_EXECUTION_STATUS_SUCCEEDED, finished.GetStatus()) + assert.Empty(t, finished.GetError()) +} + +// TestExecutionHelper_GetSecretsEmitsErroredOnFailure verifies that a failed +// vault get-secrets call finishes as errored with the returned error, so the +// failure is visible in the CRE UI. +// +// NewObserver mutates process-global beholder state, so this test must not be parallel. +func TestExecutionHelper_GetSecretsEmitsErroredOnFailure(t *testing.T) { + observer := beholdertest.NewObserver(t) + + wantErr := errors.New("vault unavailable") + h := newGetSecretsTestHelper(t, "exec-1", stubSecretsFetcher{err: wantErr}) + + _, err := h.GetSecrets(t.Context(), &sdkpb.GetSecretsRequest{CallbackId: 1}) + require.ErrorIs(t, err, wantErr) + + finished := decodeCapabilityExecutionFinished(t, observer) + assert.Equal(t, eventsv2.ExecutionStatus_EXECUTION_STATUS_FAILED, finished.GetStatus()) + assert.Equal(t, wantErr.Error(), finished.GetError()) +} From ba9794246ce54fa73cef9334dd1989c22110c533 Mon Sep 17 00:00:00 2001 From: De Clercq Wentzel <10665586+wentzeld@users.noreply.github.com> Date: Wed, 21 Oct 2026 09:16:39 -0700 Subject: [PATCH 2/6] Cover confidential workflows in vault get-secrets UI event Emit the capability execution Started/Finished pair from executionHelperWithRawSecrets.GetRawSecretsResponse as well as ExecutionHelper.GetSecrets, via a shared emitVaultGetSecrets helper, so confidential (TEE) workflows surface in the CRE UI like normal ones. GetRawSecrets delegates to GetRawSecretsResponse, so it emits once. --- .../workflows/v2/capability_executor.go | 43 ++++++--- .../v2/capability_executor_getsecrets_test.go | 89 ++++++++++++++++--- 2 files changed, 105 insertions(+), 27 deletions(-) diff --git a/core/services/workflows/v2/capability_executor.go b/core/services/workflows/v2/capability_executor.go index 7e0eaf6beb3..e90cff52f3e 100644 --- a/core/services/workflows/v2/capability_executor.go +++ b/core/services/workflows/v2/capability_executor.go @@ -301,32 +301,43 @@ func (c *ExecutionHelper) callCapability(ctx context.Context, request *sdkpb.Cap }, nil } -// GetSecrets wraps the embedded SecretsFetcher so a Vault get-secrets call is -// observable in the CRE UI as a capability execution step, matching the events -// the engine already emits for other capability calls (see callCapability). -// One Started/Finished pair is emitted per batch call; a failed call finishes as +// emitVaultGetSecrets emits a Started/Finished pair around a Vault get-secrets +// call so it surfaces in the CRE UI like any other capability call (see +// callCapability). One pair is emitted per batch call; a failed call finishes as // errored with the returned error. -func (c *ExecutionHelper) GetSecrets(ctx context.Context, request *sdkpb.GetSecretsRequest) ([]*sdkpb.SecretResponse, error) { - stepRef := strconv.FormatInt(int64(request.GetCallbackId()), 10) - labels := c.base.eventLabels() - +func emitVaultGetSecrets[T any]( + ctx context.Context, + labels map[string]string, + executionID string, + callbackID int32, + run func() (T, error), +) (T, error) { + stepRef := strconv.FormatInt(int64(callbackID), 10) _ = events.EmitCapabilityStartedEvent( - ctx, labels, c.WorkflowExecutionID, + ctx, labels, executionID, vaultcommon.CapabilityID, stepRef, vaultcommon.MethodGetSecrets, ) - resp, err := c.SecretsFetcher.GetSecrets(ctx, request) + v, err := run() status := store.StatusCompleted if err != nil { status = store.StatusErrored } _ = events.EmitCapabilityFinishedEvent( - ctx, labels, c.WorkflowExecutionID, + ctx, labels, executionID, vaultcommon.CapabilityID, stepRef, status, vaultcommon.MethodGetSecrets, err, ) + return v, err +} - return resp, err +// GetSecrets wraps the embedded SecretsFetcher so a Vault get-secrets call from a +// normal workflow is observable in the CRE UI as a capability execution step. +func (c *ExecutionHelper) GetSecrets(ctx context.Context, request *sdkpb.GetSecretsRequest) ([]*sdkpb.SecretResponse, error) { + return emitVaultGetSecrets(ctx, c.base.eventLabels(), c.WorkflowExecutionID, request.GetCallbackId(), + func() ([]*sdkpb.SecretResponse, error) { + return c.SecretsFetcher.GetSecrets(ctx, request) + }) } func (c *ExecutionHelper) GetWorkflowExecutionID() string { @@ -410,8 +421,14 @@ func (e *executionHelperWithRawSecrets) GetRawSecrets(ctx context.Context, reque return resp.GetResponses(), nil } +// GetRawSecretsResponse wraps the raw-secrets fetch so a Vault get-secrets call +// from a confidential (TEE) workflow is observable in the CRE UI as a capability +// execution step, matching the normal GetSecrets path. func (e *executionHelperWithRawSecrets) GetRawSecretsResponse(ctx context.Context, request *sdkpb.GetSecretsRequest, fetcher host.EncryptionKeyFetcher) (*vaultcommon.GetSecretsResponse, error) { - return e.SecretsFetcher.(RawSecretsFetcher).GetRawSecretsResponse(ctx, request, fetcher) + return emitVaultGetSecrets(ctx, e.base.eventLabels(), e.WorkflowExecutionID, request.GetCallbackId(), + func() (*vaultcommon.GetSecretsResponse, error) { + return e.SecretsFetcher.(RawSecretsFetcher).GetRawSecretsResponse(ctx, request, fetcher) + }) } func (e *executionHelperWithRawSecrets) GetOwner() string { diff --git a/core/services/workflows/v2/capability_executor_getsecrets_test.go b/core/services/workflows/v2/capability_executor_getsecrets_test.go index 34ad1aeb9cd..c2703514970 100644 --- a/core/services/workflows/v2/capability_executor_getsecrets_test.go +++ b/core/services/workflows/v2/capability_executor_getsecrets_test.go @@ -12,6 +12,7 @@ import ( "github.com/smartcontractkit/chainlink-common/pkg/beholder" "github.com/smartcontractkit/chainlink-common/pkg/beholder/beholdertest" vaultcommon "github.com/smartcontractkit/chainlink-common/pkg/capabilities/actions/vault" + "github.com/smartcontractkit/chainlink-common/pkg/workflows/host" sdkpb "github.com/smartcontractkit/chainlink-protos/cre/go/sdk" eventsv2 "github.com/smartcontractkit/chainlink-protos/workflows/go/v2" "github.com/smartcontractkit/chainlink/v2/core/platform" @@ -22,19 +23,28 @@ const ( entityCapabilityExecutionFinishedV2 = "workflows.v2.CapabilityExecutionFinished" ) -// stubSecretsFetcher is a minimal SecretsFetcher for exercising ExecutionHelper.GetSecrets. +// stubSecretsFetcher is a minimal SecretsFetcher/RawSecretsFetcher for exercising +// ExecutionHelper.GetSecrets and executionHelperWithRawSecrets.GetRawSecretsResponse. type stubSecretsFetcher struct { - resp []*sdkpb.SecretResponse - err error + resp []*sdkpb.SecretResponse + err error + rawResp *vaultcommon.GetSecretsResponse + rawErr error } func (s stubSecretsFetcher) GetSecrets(context.Context, *sdkpb.GetSecretsRequest) ([]*sdkpb.SecretResponse, error) { return s.resp, s.err } -// newGetSecretsTestHelper builds an ExecutionHelper whose base engine only carries -// the workflow event labels, which is all GetSecrets reads from it. -func newGetSecretsTestHelper(t *testing.T, executionID string, fetcher SecretsFetcher) *ExecutionHelper { +func (s stubSecretsFetcher) GetRawSecretsResponse(context.Context, *sdkpb.GetSecretsRequest, host.EncryptionKeyFetcher) (*vaultcommon.GetSecretsResponse, error) { + return s.rawResp, s.rawErr +} + +func (s stubSecretsFetcher) GetOwner() string { return "stub-owner" } + +// newSecretsEventTestHelper builds an ExecutionHelper whose base engine only +// carries the workflow event labels, which is all the wrappers read from it. +func newSecretsEventTestHelper(t *testing.T, executionID string, fetcher SecretsFetcher) *ExecutionHelper { t.Helper() labels := map[string]string{ @@ -71,9 +81,9 @@ func decodeCapabilityExecutionFinished(t *testing.T, observer beholdertest.Obser } // TestExecutionHelper_GetSecretsEmitsCapabilityExecutionEvents verifies that a -// successful vault get-secrets call emits a Started/Finished pair keyed to the -// real workflow execution ID, so it surfaces in the CRE UI like any other -// capability call. +// successful vault get-secrets call from a normal workflow emits a Started/Finished +// pair keyed to the real workflow execution ID, so it surfaces in the CRE UI like +// any other capability call. // // NewObserver mutates process-global beholder state, so this test must not be parallel. func TestExecutionHelper_GetSecretsEmitsCapabilityExecutionEvents(t *testing.T) { @@ -81,7 +91,7 @@ func TestExecutionHelper_GetSecretsEmitsCapabilityExecutionEvents(t *testing.T) const executionID = "exec-1" callbackID := int32(7) - h := newGetSecretsTestHelper(t, executionID, stubSecretsFetcher{}) + h := newSecretsEventTestHelper(t, executionID, stubSecretsFetcher{}) _, err := h.GetSecrets(t.Context(), &sdkpb.GetSecretsRequest{CallbackId: callbackID}) require.NoError(t, err) @@ -105,16 +115,16 @@ func TestExecutionHelper_GetSecretsEmitsCapabilityExecutionEvents(t *testing.T) assert.Empty(t, finished.GetError()) } -// TestExecutionHelper_GetSecretsEmitsErroredOnFailure verifies that a failed -// vault get-secrets call finishes as errored with the returned error, so the -// failure is visible in the CRE UI. +// TestExecutionHelper_GetSecretsEmitsErroredOnFailure verifies that a failed vault +// get-secrets call finishes as errored with the returned error, so the failure is +// visible in the CRE UI. // // NewObserver mutates process-global beholder state, so this test must not be parallel. func TestExecutionHelper_GetSecretsEmitsErroredOnFailure(t *testing.T) { observer := beholdertest.NewObserver(t) wantErr := errors.New("vault unavailable") - h := newGetSecretsTestHelper(t, "exec-1", stubSecretsFetcher{err: wantErr}) + h := newSecretsEventTestHelper(t, "exec-1", stubSecretsFetcher{err: wantErr}) _, err := h.GetSecrets(t.Context(), &sdkpb.GetSecretsRequest{CallbackId: 1}) require.ErrorIs(t, err, wantErr) @@ -123,3 +133,54 @@ func TestExecutionHelper_GetSecretsEmitsErroredOnFailure(t *testing.T) { assert.Equal(t, eventsv2.ExecutionStatus_EXECUTION_STATUS_FAILED, finished.GetStatus()) assert.Equal(t, wantErr.Error(), finished.GetError()) } + +// TestExecutionHelper_GetRawSecretsResponseEmitsCapabilityExecutionEvents verifies +// that a successful vault get-secrets call from a confidential (TEE) workflow — the +// path the confidential relay handler drives — emits the same Started/Finished pair, +// so confidential workflows surface in the CRE UI too. +// +// NewObserver mutates process-global beholder state, so this test must not be parallel. +func TestExecutionHelper_GetRawSecretsResponseEmitsCapabilityExecutionEvents(t *testing.T) { + observer := beholdertest.NewObserver(t) + + const executionID = "exec-tee" + callbackID := int32(3) + base := newSecretsEventTestHelper(t, executionID, stubSecretsFetcher{}) + raw := &executionHelperWithRawSecrets{base} + + _, err := raw.GetRawSecretsResponse(t.Context(), &sdkpb.GetSecretsRequest{CallbackId: callbackID}, nil) + require.NoError(t, err) + + started := decodeCapabilityExecutionStarted(t, observer) + assert.Equal(t, vaultcommon.CapabilityID, started.GetCapabilityID()) + assert.Equal(t, vaultcommon.MethodGetSecrets, started.GetMethod()) + assert.Equal(t, callbackID, started.GetStepRef()) + assert.Equal(t, executionID, started.GetWorkflowExecutionID()) + + finished := decodeCapabilityExecutionFinished(t, observer) + assert.Equal(t, vaultcommon.CapabilityID, finished.GetCapabilityID()) + assert.Equal(t, vaultcommon.MethodGetSecrets, finished.GetMethod()) + assert.Equal(t, callbackID, finished.GetStepRef()) + assert.Equal(t, executionID, finished.GetWorkflowExecutionID()) + assert.Equal(t, eventsv2.ExecutionStatus_EXECUTION_STATUS_SUCCEEDED, finished.GetStatus()) + assert.Empty(t, finished.GetError()) +} + +// TestExecutionHelper_GetRawSecretsResponseEmitsErroredOnFailure verifies that a +// failed confidential get-secrets call finishes as errored with the returned error. +// +// NewObserver mutates process-global beholder state, so this test must not be parallel. +func TestExecutionHelper_GetRawSecretsResponseEmitsErroredOnFailure(t *testing.T) { + observer := beholdertest.NewObserver(t) + + wantErr := errors.New("vault unavailable") + base := newSecretsEventTestHelper(t, "exec-tee", stubSecretsFetcher{rawErr: wantErr}) + raw := &executionHelperWithRawSecrets{base} + + _, err := raw.GetRawSecretsResponse(t.Context(), &sdkpb.GetSecretsRequest{CallbackId: 1}, nil) + require.ErrorIs(t, err, wantErr) + + finished := decodeCapabilityExecutionFinished(t, observer) + assert.Equal(t, eventsv2.ExecutionStatus_EXECUTION_STATUS_FAILED, finished.GetStatus()) + assert.Equal(t, wantErr.Error(), finished.GetError()) +} From f7f1ca9a6efd673c9bd0996a30f9e447734b6a15 Mon Sep 17 00:00:00 2001 From: De Clercq Wentzel <10665586+wentzeld@users.noreply.github.com> Date: Wed, 21 Oct 2026 09:33:08 -0700 Subject: [PATCH 3/6] Silence paralleltest for global beholder observer tests These tests can't call t.Parallel(): they assert on events emitted via beholder.GetEmitter(), the process-global client, so they must swap the global client with beholdertest.NewObserver, which is not parallel-safe. Matches the existing convention (e.g. engine_drop_paths_test.go). --- .../workflows/v2/capability_executor_getsecrets_test.go | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/core/services/workflows/v2/capability_executor_getsecrets_test.go b/core/services/workflows/v2/capability_executor_getsecrets_test.go index c2703514970..33bfa296dcf 100644 --- a/core/services/workflows/v2/capability_executor_getsecrets_test.go +++ b/core/services/workflows/v2/capability_executor_getsecrets_test.go @@ -86,7 +86,7 @@ func decodeCapabilityExecutionFinished(t *testing.T, observer beholdertest.Obser // any other capability call. // // NewObserver mutates process-global beholder state, so this test must not be parallel. -func TestExecutionHelper_GetSecretsEmitsCapabilityExecutionEvents(t *testing.T) { +func TestExecutionHelper_GetSecretsEmitsCapabilityExecutionEvents(t *testing.T) { //nolint:paralleltest // uses beholdertest.NewObserver, a global singleton swap observer := beholdertest.NewObserver(t) const executionID = "exec-1" @@ -120,7 +120,7 @@ func TestExecutionHelper_GetSecretsEmitsCapabilityExecutionEvents(t *testing.T) // visible in the CRE UI. // // NewObserver mutates process-global beholder state, so this test must not be parallel. -func TestExecutionHelper_GetSecretsEmitsErroredOnFailure(t *testing.T) { +func TestExecutionHelper_GetSecretsEmitsErroredOnFailure(t *testing.T) { //nolint:paralleltest // uses beholdertest.NewObserver, a global singleton swap observer := beholdertest.NewObserver(t) wantErr := errors.New("vault unavailable") @@ -140,7 +140,7 @@ func TestExecutionHelper_GetSecretsEmitsErroredOnFailure(t *testing.T) { // so confidential workflows surface in the CRE UI too. // // NewObserver mutates process-global beholder state, so this test must not be parallel. -func TestExecutionHelper_GetRawSecretsResponseEmitsCapabilityExecutionEvents(t *testing.T) { +func TestExecutionHelper_GetRawSecretsResponseEmitsCapabilityExecutionEvents(t *testing.T) { //nolint:paralleltest // uses beholdertest.NewObserver, a global singleton swap observer := beholdertest.NewObserver(t) const executionID = "exec-tee" @@ -170,7 +170,7 @@ func TestExecutionHelper_GetRawSecretsResponseEmitsCapabilityExecutionEvents(t * // failed confidential get-secrets call finishes as errored with the returned error. // // NewObserver mutates process-global beholder state, so this test must not be parallel. -func TestExecutionHelper_GetRawSecretsResponseEmitsErroredOnFailure(t *testing.T) { +func TestExecutionHelper_GetRawSecretsResponseEmitsErroredOnFailure(t *testing.T) { //nolint:paralleltest // uses beholdertest.NewObserver, a global singleton swap observer := beholdertest.NewObserver(t) wantErr := errors.New("vault unavailable") From b7ff1d274137b151ecb573dea518b298a0f70eed Mon Sep 17 00:00:00 2001 From: De Clercq Wentzel <10665586+wentzeld@users.noreply.github.com> Date: Fri, 23 Oct 2026 12:58:57 -0700 Subject: [PATCH 4/6] Report partial vault get-secrets batches as completed-with-errors Classify each get-secrets batch by per-secret outcome: all succeed -> completed, some fail -> completed with an error summary (so the CRE UI derives completed_with_errors), all fail -> errored. The call still returns its responses; only the emitted event status/error changes. --- .../workflows/v2/capability_executor.go | 67 +++++++++-- .../v2/capability_executor_getsecrets_test.go | 112 ++++++++++++++++-- 2 files changed, 160 insertions(+), 19 deletions(-) diff --git a/core/services/workflows/v2/capability_executor.go b/core/services/workflows/v2/capability_executor.go index e90cff52f3e..ace7fa0ae4b 100644 --- a/core/services/workflows/v2/capability_executor.go +++ b/core/services/workflows/v2/capability_executor.go @@ -301,16 +301,23 @@ func (c *ExecutionHelper) callCapability(ctx context.Context, request *sdkpb.Cap }, nil } -// emitVaultGetSecrets emits a Started/Finished pair around a Vault get-secrets -// call so it surfaces in the CRE UI like any other capability call (see -// callCapability). One pair is emitted per batch call; a failed call finishes as -// errored with the returned error. -func emitVaultGetSecrets[T any]( +// runWithGetSecretsEvents runs fn (a Vault get-secrets call) and emits a +// Started/Finished pair around it so the call surfaces in the CRE UI like any +// other capability call (see callCapability). One pair is emitted per batch +// call. A call-level error finishes as errored; otherwise outcome classifies the +// per-secret result — a partial batch finishes succeeded with an error summary, +// which the CRE UI uses to derive "completed with errors" for the execution. +// +// The returned error is the call-level error only: a partial or fully-failed +// batch still returns its responses to the caller (the per-secret errors live in +// them). +func runWithGetSecretsEvents[T any]( ctx context.Context, labels map[string]string, executionID string, callbackID int32, run func() (T, error), + outcome func(T) (string, error), ) (T, error) { stepRef := strconv.FormatInt(int64(callbackID), 10) _ = events.EmitCapabilityStartedEvent( @@ -320,23 +327,51 @@ func emitVaultGetSecrets[T any]( v, err := run() - status := store.StatusCompleted - if err != nil { - status = store.StatusErrored + emitStatus := store.StatusCompleted + var emitErr error + switch { + case err != nil: + emitStatus, emitErr = store.StatusErrored, err + case outcome != nil: + emitStatus, emitErr = outcome(v) } _ = events.EmitCapabilityFinishedEvent( ctx, labels, executionID, - vaultcommon.CapabilityID, stepRef, status, vaultcommon.MethodGetSecrets, err, + vaultcommon.CapabilityID, stepRef, emitStatus, vaultcommon.MethodGetSecrets, emitErr, ) return v, err } +// vaultSecretBatchOutcome classifies a get-secrets batch for the CRE UI event: +// all requests succeeded, some failed (partial), or all failed. The summary is +// counts only — never secret keys or values — so the event carries no +// secret-identifying data. +func vaultSecretBatchOutcome(total, failed int) (string, error) { + switch { + case failed == 0: + return store.StatusCompleted, nil + case failed == total: + return store.StatusErrored, fmt.Errorf("all %d secret requests failed", total) + default: + return store.StatusCompleted, fmt.Errorf("%d of %d secret requests failed", failed, total) + } +} + // GetSecrets wraps the embedded SecretsFetcher so a Vault get-secrets call from a // normal workflow is observable in the CRE UI as a capability execution step. func (c *ExecutionHelper) GetSecrets(ctx context.Context, request *sdkpb.GetSecretsRequest) ([]*sdkpb.SecretResponse, error) { - return emitVaultGetSecrets(ctx, c.base.eventLabels(), c.WorkflowExecutionID, request.GetCallbackId(), + return runWithGetSecretsEvents(ctx, c.base.eventLabels(), c.WorkflowExecutionID, request.GetCallbackId(), func() ([]*sdkpb.SecretResponse, error) { return c.SecretsFetcher.GetSecrets(ctx, request) + }, + func(responses []*sdkpb.SecretResponse) (string, error) { + failed := 0 + for _, r := range responses { + if r.GetError() != nil { + failed++ + } + } + return vaultSecretBatchOutcome(len(responses), failed) }) } @@ -425,9 +460,19 @@ func (e *executionHelperWithRawSecrets) GetRawSecrets(ctx context.Context, reque // from a confidential (TEE) workflow is observable in the CRE UI as a capability // execution step, matching the normal GetSecrets path. func (e *executionHelperWithRawSecrets) GetRawSecretsResponse(ctx context.Context, request *sdkpb.GetSecretsRequest, fetcher host.EncryptionKeyFetcher) (*vaultcommon.GetSecretsResponse, error) { - return emitVaultGetSecrets(ctx, e.base.eventLabels(), e.WorkflowExecutionID, request.GetCallbackId(), + return runWithGetSecretsEvents(ctx, e.base.eventLabels(), e.WorkflowExecutionID, request.GetCallbackId(), func() (*vaultcommon.GetSecretsResponse, error) { return e.SecretsFetcher.(RawSecretsFetcher).GetRawSecretsResponse(ctx, request, fetcher) + }, + func(resp *vaultcommon.GetSecretsResponse) (string, error) { + responses := resp.GetResponses() + failed := 0 + for _, r := range responses { + if r.GetError() != "" { + failed++ + } + } + return vaultSecretBatchOutcome(len(responses), failed) }) } diff --git a/core/services/workflows/v2/capability_executor_getsecrets_test.go b/core/services/workflows/v2/capability_executor_getsecrets_test.go index 33bfa296dcf..a21b04727e7 100644 --- a/core/services/workflows/v2/capability_executor_getsecrets_test.go +++ b/core/services/workflows/v2/capability_executor_getsecrets_test.go @@ -42,6 +42,32 @@ func (s stubSecretsFetcher) GetRawSecretsResponse(context.Context, *sdkpb.GetSec func (s stubSecretsFetcher) GetOwner() string { return "stub-owner" } +func sdkSecretOK(id string) *sdkpb.SecretResponse { + return &sdkpb.SecretResponse{ + Response: &sdkpb.SecretResponse_Secret{Secret: &sdkpb.Secret{Id: id, Value: "value-" + id}}, + } +} + +func sdkSecretErr(id string) *sdkpb.SecretResponse { + return &sdkpb.SecretResponse{ + Response: &sdkpb.SecretResponse_Error{Error: &sdkpb.SecretError{Id: id, Error: "not found"}}, + } +} + +func vaultSecretOK(id string) *vaultcommon.SecretResponse { + return &vaultcommon.SecretResponse{ + Id: &vaultcommon.SecretIdentifier{Key: id}, + Result: &vaultcommon.SecretResponse_Data{Data: &vaultcommon.SecretData{}}, + } +} + +func vaultSecretErr(id string) *vaultcommon.SecretResponse { + return &vaultcommon.SecretResponse{ + Id: &vaultcommon.SecretIdentifier{Key: id}, + Result: &vaultcommon.SecretResponse_Error{Error: "not found"}, + } +} + // newSecretsEventTestHelper builds an ExecutionHelper whose base engine only // carries the workflow event labels, which is all the wrappers read from it. func newSecretsEventTestHelper(t *testing.T, executionID string, fetcher SecretsFetcher) *ExecutionHelper { @@ -91,7 +117,9 @@ func TestExecutionHelper_GetSecretsEmitsCapabilityExecutionEvents(t *testing.T) const executionID = "exec-1" callbackID := int32(7) - h := newSecretsEventTestHelper(t, executionID, stubSecretsFetcher{}) + h := newSecretsEventTestHelper(t, executionID, stubSecretsFetcher{ + resp: []*sdkpb.SecretResponse{sdkSecretOK("a"), sdkSecretOK("b")}, + }) _, err := h.GetSecrets(t.Context(), &sdkpb.GetSecretsRequest{CallbackId: callbackID}) require.NoError(t, err) @@ -115,12 +143,12 @@ func TestExecutionHelper_GetSecretsEmitsCapabilityExecutionEvents(t *testing.T) assert.Empty(t, finished.GetError()) } -// TestExecutionHelper_GetSecretsEmitsErroredOnFailure verifies that a failed vault -// get-secrets call finishes as errored with the returned error, so the failure is -// visible in the CRE UI. +// TestExecutionHelper_GetSecretsEmitsErroredOnCallFailure verifies that a +// call-level failure finishes as errored with the returned error, so the failure +// is visible in the CRE UI. // // NewObserver mutates process-global beholder state, so this test must not be parallel. -func TestExecutionHelper_GetSecretsEmitsErroredOnFailure(t *testing.T) { //nolint:paralleltest // uses beholdertest.NewObserver, a global singleton swap +func TestExecutionHelper_GetSecretsEmitsErroredOnCallFailure(t *testing.T) { //nolint:paralleltest // uses beholdertest.NewObserver, a global singleton swap observer := beholdertest.NewObserver(t) wantErr := errors.New("vault unavailable") @@ -134,6 +162,48 @@ func TestExecutionHelper_GetSecretsEmitsErroredOnFailure(t *testing.T) { //nolin assert.Equal(t, wantErr.Error(), finished.GetError()) } +// TestExecutionHelper_GetSecretsEmitsPartialOutcome verifies that a batch where +// only some secrets fail finishes succeeded with an error summary, so the +// workflow execution derives "completed with errors" in the CRE UI. +// +// NewObserver mutates process-global beholder state, so this test must not be parallel. +func TestExecutionHelper_GetSecretsEmitsPartialOutcome(t *testing.T) { //nolint:paralleltest // uses beholdertest.NewObserver, a global singleton swap + observer := beholdertest.NewObserver(t) + + h := newSecretsEventTestHelper(t, "exec-1", stubSecretsFetcher{ + resp: []*sdkpb.SecretResponse{sdkSecretOK("a"), sdkSecretOK("b"), sdkSecretErr("c")}, + }) + + resp, err := h.GetSecrets(t.Context(), &sdkpb.GetSecretsRequest{CallbackId: 1}) + require.NoError(t, err, "a partial batch is still a successful call") + require.Len(t, resp, 3) + + finished := decodeCapabilityExecutionFinished(t, observer) + assert.Equal(t, eventsv2.ExecutionStatus_EXECUTION_STATUS_SUCCEEDED, finished.GetStatus()) + assert.Equal(t, "1 of 3 secret requests failed", finished.GetError()) +} + +// TestExecutionHelper_GetSecretsEmitsErroredWhenAllSecretsFail verifies that a +// batch where every secret fails finishes errored (the call itself still returns +// its responses), so the step is visibly failed in the CRE UI. +// +// NewObserver mutates process-global beholder state, so this test must not be parallel. +func TestExecutionHelper_GetSecretsEmitsErroredWhenAllSecretsFail(t *testing.T) { //nolint:paralleltest // uses beholdertest.NewObserver, a global singleton swap + observer := beholdertest.NewObserver(t) + + h := newSecretsEventTestHelper(t, "exec-1", stubSecretsFetcher{ + resp: []*sdkpb.SecretResponse{sdkSecretErr("a"), sdkSecretErr("b")}, + }) + + resp, err := h.GetSecrets(t.Context(), &sdkpb.GetSecretsRequest{CallbackId: 1}) + require.NoError(t, err, "per-secret failures are not a call-level error") + require.Len(t, resp, 2) + + finished := decodeCapabilityExecutionFinished(t, observer) + assert.Equal(t, eventsv2.ExecutionStatus_EXECUTION_STATUS_FAILED, finished.GetStatus()) + assert.Equal(t, "all 2 secret requests failed", finished.GetError()) +} + // TestExecutionHelper_GetRawSecretsResponseEmitsCapabilityExecutionEvents verifies // that a successful vault get-secrets call from a confidential (TEE) workflow — the // path the confidential relay handler drives — emits the same Started/Finished pair, @@ -145,7 +215,11 @@ func TestExecutionHelper_GetRawSecretsResponseEmitsCapabilityExecutionEvents(t * const executionID = "exec-tee" callbackID := int32(3) - base := newSecretsEventTestHelper(t, executionID, stubSecretsFetcher{}) + base := newSecretsEventTestHelper(t, executionID, stubSecretsFetcher{ + rawResp: &vaultcommon.GetSecretsResponse{ + Responses: []*vaultcommon.SecretResponse{vaultSecretOK("a")}, + }, + }) raw := &executionHelperWithRawSecrets{base} _, err := raw.GetRawSecretsResponse(t.Context(), &sdkpb.GetSecretsRequest{CallbackId: callbackID}, nil) @@ -166,11 +240,33 @@ func TestExecutionHelper_GetRawSecretsResponseEmitsCapabilityExecutionEvents(t * assert.Empty(t, finished.GetError()) } -// TestExecutionHelper_GetRawSecretsResponseEmitsErroredOnFailure verifies that a +// TestExecutionHelper_GetRawSecretsResponseEmitsPartialOutcome verifies the +// confidential path classifies a partial batch the same way as the normal path. +// +// NewObserver mutates process-global beholder state, so this test must not be parallel. +func TestExecutionHelper_GetRawSecretsResponseEmitsPartialOutcome(t *testing.T) { //nolint:paralleltest // uses beholdertest.NewObserver, a global singleton swap + observer := beholdertest.NewObserver(t) + + base := newSecretsEventTestHelper(t, "exec-tee", stubSecretsFetcher{ + rawResp: &vaultcommon.GetSecretsResponse{ + Responses: []*vaultcommon.SecretResponse{vaultSecretOK("a"), vaultSecretErr("b")}, + }, + }) + raw := &executionHelperWithRawSecrets{base} + + _, err := raw.GetRawSecretsResponse(t.Context(), &sdkpb.GetSecretsRequest{CallbackId: 1}, nil) + require.NoError(t, err) + + finished := decodeCapabilityExecutionFinished(t, observer) + assert.Equal(t, eventsv2.ExecutionStatus_EXECUTION_STATUS_SUCCEEDED, finished.GetStatus()) + assert.Equal(t, "1 of 2 secret requests failed", finished.GetError()) +} + +// TestExecutionHelper_GetRawSecretsResponseEmitsErroredOnCallFailure verifies that a // failed confidential get-secrets call finishes as errored with the returned error. // // NewObserver mutates process-global beholder state, so this test must not be parallel. -func TestExecutionHelper_GetRawSecretsResponseEmitsErroredOnFailure(t *testing.T) { //nolint:paralleltest // uses beholdertest.NewObserver, a global singleton swap +func TestExecutionHelper_GetRawSecretsResponseEmitsErroredOnCallFailure(t *testing.T) { //nolint:paralleltest // uses beholdertest.NewObserver, a global singleton swap observer := beholdertest.NewObserver(t) wantErr := errors.New("vault unavailable") From f392a904f515f3ad14841bd36bffeae2246cc8a5 Mon Sep 17 00:00:00 2001 From: De Clercq Wentzel <10665586+wentzeld@users.noreply.github.com> Date: Fri, 23 Oct 2026 16:26:14 -0700 Subject: [PATCH 5/6] Emit vault get-secrets events via injected EventEmitter --- .../workflows/v2/capability_executor.go | 25 ++++++++++--------- .../v2/capability_executor_getsecrets_test.go | 3 ++- 2 files changed, 15 insertions(+), 13 deletions(-) diff --git a/core/services/workflows/v2/capability_executor.go b/core/services/workflows/v2/capability_executor.go index ace7fa0ae4b..3e2fca992f7 100644 --- a/core/services/workflows/v2/capability_executor.go +++ b/core/services/workflows/v2/capability_executor.go @@ -302,26 +302,27 @@ func (c *ExecutionHelper) callCapability(ctx context.Context, request *sdkpb.Cap } // runWithGetSecretsEvents runs fn (a Vault get-secrets call) and emits a -// Started/Finished pair around it so the call surfaces in the CRE UI like any -// other capability call (see callCapability). One pair is emitted per batch -// call. A call-level error finishes as errored; otherwise outcome classifies the -// per-secret result — a partial batch finishes succeeded with an error summary, -// which the CRE UI uses to derive "completed with errors" for the execution. +// Started/Finished pair around it via the engine's injected event emitter, so +// the call surfaces in the CRE UI like any other capability call (see +// callCapability). One pair is emitted per batch call. A call-level error +// finishes as errored; otherwise outcome classifies the per-secret result — a +// partial batch finishes succeeded with an error summary, which the CRE UI uses +// to derive "completed with errors" for the execution. // // The returned error is the call-level error only: a partial or fully-failed // batch still returns its responses to the caller (the per-secret errors live in // them). func runWithGetSecretsEvents[T any]( ctx context.Context, - labels map[string]string, + base *baseEngine, executionID string, callbackID int32, run func() (T, error), outcome func(T) (string, error), ) (T, error) { stepRef := strconv.FormatInt(int64(callbackID), 10) - _ = events.EmitCapabilityStartedEvent( - ctx, labels, executionID, + _ = base.cfg.EventEmitter.EmitCapabilityStartedEvent( + ctx, base.eventLabels(), executionID, vaultcommon.CapabilityID, stepRef, vaultcommon.MethodGetSecrets, ) @@ -335,8 +336,8 @@ func runWithGetSecretsEvents[T any]( case outcome != nil: emitStatus, emitErr = outcome(v) } - _ = events.EmitCapabilityFinishedEvent( - ctx, labels, executionID, + _ = base.cfg.EventEmitter.EmitCapabilityFinishedEvent( + ctx, base.eventLabels(), executionID, vaultcommon.CapabilityID, stepRef, emitStatus, vaultcommon.MethodGetSecrets, emitErr, ) return v, err @@ -360,7 +361,7 @@ func vaultSecretBatchOutcome(total, failed int) (string, error) { // GetSecrets wraps the embedded SecretsFetcher so a Vault get-secrets call from a // normal workflow is observable in the CRE UI as a capability execution step. func (c *ExecutionHelper) GetSecrets(ctx context.Context, request *sdkpb.GetSecretsRequest) ([]*sdkpb.SecretResponse, error) { - return runWithGetSecretsEvents(ctx, c.base.eventLabels(), c.WorkflowExecutionID, request.GetCallbackId(), + return runWithGetSecretsEvents(ctx, c.base, c.WorkflowExecutionID, request.GetCallbackId(), func() ([]*sdkpb.SecretResponse, error) { return c.SecretsFetcher.GetSecrets(ctx, request) }, @@ -460,7 +461,7 @@ func (e *executionHelperWithRawSecrets) GetRawSecrets(ctx context.Context, reque // from a confidential (TEE) workflow is observable in the CRE UI as a capability // execution step, matching the normal GetSecrets path. func (e *executionHelperWithRawSecrets) GetRawSecretsResponse(ctx context.Context, request *sdkpb.GetSecretsRequest, fetcher host.EncryptionKeyFetcher) (*vaultcommon.GetSecretsResponse, error) { - return runWithGetSecretsEvents(ctx, e.base.eventLabels(), e.WorkflowExecutionID, request.GetCallbackId(), + return runWithGetSecretsEvents(ctx, e.base, e.WorkflowExecutionID, request.GetCallbackId(), func() (*vaultcommon.GetSecretsResponse, error) { return e.SecretsFetcher.(RawSecretsFetcher).GetRawSecretsResponse(ctx, request, fetcher) }, diff --git a/core/services/workflows/v2/capability_executor_getsecrets_test.go b/core/services/workflows/v2/capability_executor_getsecrets_test.go index a21b04727e7..296df6fd818 100644 --- a/core/services/workflows/v2/capability_executor_getsecrets_test.go +++ b/core/services/workflows/v2/capability_executor_getsecrets_test.go @@ -16,6 +16,7 @@ import ( sdkpb "github.com/smartcontractkit/chainlink-protos/cre/go/sdk" eventsv2 "github.com/smartcontractkit/chainlink-protos/workflows/go/v2" "github.com/smartcontractkit/chainlink/v2/core/platform" + "github.com/smartcontractkit/chainlink/v2/core/services/workflows/events" ) const ( @@ -78,7 +79,7 @@ func newSecretsEventTestHelper(t *testing.T, executionID string, fetcher Secrets platform.KeyWorkflowID: "wf-id", platform.KeyWorkflowName: "wf-name", } - be := &baseEngine{} + be := &baseEngine{cfg: &EngineConfig{EventEmitter: events.DefaultEmitter()}} be.loggerLabels.Store(&labels) return &ExecutionHelper{ From 47deb2e9b91aeb6b5d3d293629799e4bb03ec657 Mon Sep 17 00:00:00 2001 From: De Clercq Wentzel <10665586+wentzeld@users.noreply.github.com> Date: Fri, 23 Oct 2026 16:31:10 -0700 Subject: [PATCH 6/6] Use tagged switch in vaultSecretBatchOutcome (staticcheck QF1002) --- core/services/workflows/v2/capability_executor.go | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/core/services/workflows/v2/capability_executor.go b/core/services/workflows/v2/capability_executor.go index 3e2fca992f7..b2b4ef0068f 100644 --- a/core/services/workflows/v2/capability_executor.go +++ b/core/services/workflows/v2/capability_executor.go @@ -348,10 +348,10 @@ func runWithGetSecretsEvents[T any]( // counts only — never secret keys or values — so the event carries no // secret-identifying data. func vaultSecretBatchOutcome(total, failed int) (string, error) { - switch { - case failed == 0: + switch failed { + case 0: return store.StatusCompleted, nil - case failed == total: + case total: return store.StatusErrored, fmt.Errorf("all %d secret requests failed", total) default: return store.StatusCompleted, fmt.Errorf("%d of %d secret requests failed", failed, total)