Support Worker-variant callbacks

chasm/context_mock.go

@@ -46,6 +46,16 @@ type MockContext struct {
 	registeredContextValues map[any]any
 }
 
+// RegisterLibrary copies the context values that lib's components declare via
+// [WithContextValues] into the mock, mirroring what [Registry] does in production. Use this in
+// tests that reach into another library's component methods, so that library can keep its context
+// keys unexported.
+func (c *MockContext) RegisterLibrary(lib Library) {
+	for _, rc := range lib.Components() {
+		c.RegisterComponentContextValues(rc.contextValues)
+	}
+}
+
 func (c *MockContext) RegisterComponentContextValues(
 	keyValues map[any]any,
 ) {
@@ -138,7 +148,11 @@ func (c *MockContext) MetricsHandler() metrics.Handler {
 }
 
 func (c *MockContext) Value(key any) any {
-	return c.goContext().Value(key)
+	if v := c.goContext().Value(key); v != nil {
+		return v
+	}
+
+	return c.registeredContextValues[key]
 }
 
 func (c *MockContext) Links(component Component) []*commonpb.Link {
@@ -175,6 +189,8 @@ func (c *MockContext) withValue(key any, value any) Context {
 		HandleLinks:          c.HandleLinks,
 		HandleRequestLinks:   c.HandleRequestLinks,
 		HandleUserMetadata:   c.HandleUserMetadata,
+
+		registeredContextValues: c.registeredContextValues,
 	}
 }
 

chasm/lib/activity/activity.go

@@ -17,7 +17,6 @@ import (
 	"go.temporal.io/server/chasm"
 	"go.temporal.io/server/chasm/lib/activity/gen/activitypb/v1"
 	"go.temporal.io/server/chasm/lib/callback"
-	callbackspb "go.temporal.io/server/chasm/lib/callback/gen/callbackpb/v1"
 	"go.temporal.io/server/common"
 	"go.temporal.io/server/common/contextutil"
 	"go.temporal.io/server/common/metrics"
@@ -326,24 +325,14 @@ func (a *Activity) addCompletionCallbacks(
 	registrationTime := timestamppb.New(ctx.Now(a))
 
 	for idx, cb := range completionCallbacks {
-		chasmCB := &callbackspb.Callback{
-			Links: cb.GetLinks(),
-		}
-		switch variant := cb.Variant.(type) {
-		case *commonpb.Callback_Nexus_:
-			chasmCB.Variant = &callbackspb.Callback_Nexus_{
-				Nexus: &callbackspb.Callback_Nexus{
-					Url:    variant.Nexus.GetUrl(),
-					Header: variant.Nexus.GetHeader(),
-				},
-			}
-		default:
-			return serviceerror.NewInvalidArgumentf("unsupported callback variant: %T", variant)
+		chasmCB, err := callback.FromAPICallback(cb)
+		if err != nil {
+			return err
 		}
 
 		// requestID (unique per API call) + idx (position within the request) ensures unique,idempotent callback IDs.
 		id := fmt.Sprintf("%s-%d", requestID, idx)
-		callbackObj := callback.NewCallback(requestID, registrationTime, &callbackspb.CallbackState{}, chasmCB)
+		callbackObj := callback.NewCallback(requestID, registrationTime, chasmCB)
 		a.Callbacks[id] = chasm.NewComponentField(ctx, callbackObj)
 	}
 	return nil

chasm/lib/activity/config.go

@@ -3,6 +3,7 @@ package activity
 import (
 	"go.temporal.io/server/chasm/lib/callback"
 	"go.temporal.io/server/common"
+	"go.temporal.io/server/common/callbacks"
 	"go.temporal.io/server/common/dynamicconfig"
 	"go.temporal.io/server/common/retrypolicy"
 )
@@ -40,6 +41,14 @@ var (
 		`Allows attaching completion callbacks to standalone activity executions.`,
 	)
 
+	EnabledCallbackKinds = dynamicconfig.NewNamespaceTypedSettingWithConverter(
+		"activity.enabledCallbackKinds",
+		callbacks.ConvertEnabledKinds,
+		[]callbacks.Kind{callbacks.KindNexus},
+		`The list of completion callback kinds that may be attached to a standalone activity execution.
+Only consulted when activity.enableCallbacks is set.`,
+	)
+
 	EnableStandaloneActivityOperatorCommands = dynamicconfig.NewNamespaceBoolSetting(
 		"history.enableStandaloneActivityOperatorCommands",
 		false,
@@ -52,6 +61,7 @@ type Config struct {
 	BlobSizeLimitWarn                         dynamicconfig.IntPropertyFnWithNamespaceFilter
 	BreakdownMetricsByTaskQueue               dynamicconfig.TypedPropertyFnWithTaskQueueFilter[bool]
 	EnableCallbacks                           dynamicconfig.BoolPropertyFnWithNamespaceFilter
+	EnabledCallbackKinds                      dynamicconfig.TypedPropertyFnWithNamespaceFilter[[]callbacks.Kind]
 	Enabled                                   dynamicconfig.BoolPropertyFnWithNamespaceFilter
 	EnableStandaloneActivityOperatorCommands  dynamicconfig.BoolPropertyFnWithNamespaceFilter
 	LongPollBuffer                            dynamicconfig.DurationPropertyFnWithNamespaceFilter
@@ -73,6 +83,7 @@ func ConfigProvider(dc *dynamicconfig.Collection) *Config {
 		BreakdownMetricsByTaskQueue:               dynamicconfig.MetricsBreakdownByTaskQueue.Get(dc),
 		DefaultActivityRetryPolicy:                dynamicconfig.DefaultActivityRetryPolicy.Get(dc),
 		EnableCallbacks:                           EnableCallbacks.Get(dc),
+		EnabledCallbackKinds:                      EnabledCallbackKinds.Get(dc),
 		Enabled:                                   Enabled.Get(dc),
 		EnableStandaloneActivityOperatorCommands:  EnableStandaloneActivityOperatorCommands.Get(dc),
 		LongPollBuffer:                            LongPollBuffer.Get(dc),

chasm/lib/activity/frontend.go

@@ -418,7 +418,10 @@ func (h *frontendHandler) validateAndPopulateStartRequest(
 		if !h.config.EnableCallbacks(req.GetNamespace()) {
 			return nil, serviceerror.NewInvalidArgument("completion callbacks are not enabled for this namespace")
 		}
-		if err := h.callbackValidator.Validate(ctx, req.GetNamespace(), cbs); err != nil {
+		opts := callbacks.ValidatorOptions{
+			EnabledKinds: h.config.EnabledCallbackKinds(req.GetNamespace()),
+		}
+		if err := h.callbackValidator.Validate(ctx, req.GetNamespace(), cbs, opts); err != nil {
 			return nil, err
 		}
 	}

chasm/lib/activity/responses.go

@@ -4,15 +4,12 @@ import (
 	"fmt"
 
 	apiactivitypb "go.temporal.io/api/activity/v1" //nolint:importas
-	callbackpb "go.temporal.io/api/callback/v1"
 	commonpb "go.temporal.io/api/common/v1"
 	enumspb "go.temporal.io/api/enums/v1"
 	failurepb "go.temporal.io/api/failure/v1"
-	"go.temporal.io/api/serviceerror"
 	"go.temporal.io/api/workflowservice/v1"
 	"go.temporal.io/server/chasm"
 	"go.temporal.io/server/chasm/lib/activity/gen/activitypb/v1"
-	callbackspb "go.temporal.io/server/chasm/lib/callback/gen/callbackpb/v1"
 	"google.golang.org/protobuf/types/known/durationpb"
 	"google.golang.org/protobuf/types/known/timestamppb"
 )
@@ -205,42 +202,16 @@ func (a *Activity) buildCallbackInfos(ctx chasm.Context) ([]*apiactivitypb.Callb
 	for _, field := range a.Callbacks {
 		cb := field.Get(ctx)
 
-		cbSpec, err := cb.ToAPICallback()
+		cbInfo, err := cb.ToAPICallbackInfo(ctx)
 		if err != nil {
 			return nil, err
 		}
 
-		var state enumspb.CallbackState
-		switch cb.Status {
-		case callbackspb.CALLBACK_STATUS_UNSPECIFIED:
-			return nil, serviceerror.NewInternal("callback with UNSPECIFIED state")
-		case callbackspb.CALLBACK_STATUS_STANDBY:
-			state = enumspb.CALLBACK_STATE_STANDBY
-		case callbackspb.CALLBACK_STATUS_SCHEDULED:
-			state = enumspb.CALLBACK_STATE_SCHEDULED
-		case callbackspb.CALLBACK_STATUS_BACKING_OFF:
-			state = enumspb.CALLBACK_STATE_BACKING_OFF
-		case callbackspb.CALLBACK_STATUS_FAILED:
-			state = enumspb.CALLBACK_STATE_FAILED
-		case callbackspb.CALLBACK_STATUS_SUCCEEDED:
-			state = enumspb.CALLBACK_STATE_SUCCEEDED
-		default:
-			return nil, serviceerror.NewInternalf("unknown callback state: %v", cb.Status)
-		}
-
 		cbInfos = append(cbInfos, &apiactivitypb.CallbackInfo{
 			Trigger: &apiactivitypb.CallbackInfo_Trigger{
 				Variant: &apiactivitypb.CallbackInfo_Trigger_ActivityClosed{},
 			},
-			Info: &callbackpb.CallbackInfo{
-				Callback:                cbSpec,
-				RegistrationTime:        cb.RegistrationTime,
-				State:                   state,
-				Attempt:                 cb.Attempt,
-				LastAttemptCompleteTime: cb.LastAttemptCompleteTime,
-				LastAttemptFailure:      cb.LastAttemptFailure,
-				NextAttemptScheduleTime: cb.NextAttemptScheduleTime,
-			},
+			Info: cbInfo,
 		})
 	}
 	return cbInfos, nil

chasm/lib/activity/validator_test.go

@@ -536,6 +536,7 @@ func TestRequestIDGeneratedWhenMissing(t *testing.T) {
 func TestValidateAndPopulateStartRequest_CombinesRequestAndCallbackLinks(t *testing.T) {
 	callbackValidator, err := callbacks.NewValidator(callbacks.ValidatorConfig{
 		MaxCallbacksPerExecution: func(string) int { return 2000 },
+		MaxIDLengthLimit:         func() int { return 1000 },
 		URLMaxLength:             func(string) int { return 1000 },
 		HeaderMaxSize:            func(string) int { return 2000 },
 		EndpointRules: func(string) callbacks.AddressMatchRules {
@@ -545,6 +546,10 @@ func TestValidateAndPopulateStartRequest_CombinesRequestAndCallbackLinks(t *test
 				},
 			}
 		},
+		MaxServiceNameLength:                func(string) int { return 1000 },
+		MaxOperationNameLength:              func(string) int { return 1000 },
+		WorkerSourceContextMaxSize:          func(string) int { return 64 * 1024 },
+		WorkerSourceContextAggregateMaxSize: func(string) int { return 2 * 1024 * 1024 },
 	})
 	require.NoError(t, err)
 
@@ -554,6 +559,9 @@ func TestValidateAndPopulateStartRequest_CombinesRequestAndCallbackLinks(t *test
 			BlobSizeLimitWarn:          defaultBlobSizeLimitWarn,
 			DefaultActivityRetryPolicy: getDefaultRetrySettings,
 			EnableCallbacks:            func(string) bool { return true },
+			EnabledCallbackKinds: func(string) []callbacks.Kind {
+				return []callbacks.Kind{callbacks.KindNexus}
+			},
 			MaxIDLengthLimit:           func() int { return defaultMaxIDLengthLimit },
 			MaxUserMetadataDetailsSize: defaultMaxUserMetadataDetailsSize,
 			MaxUserMetadataSummarySize: defaultMaxUserMetadataSummarySize,
@@ -580,8 +588,10 @@ func TestValidateAndPopulateStartRequest_CombinesRequestAndCallbackLinks(t *test
 			},
 		}},
 		CompletionCallbacks: []*commonpb.Callback{{
-			Variant: &commonpb.Callback_Internal_{
-				Internal: &commonpb.Callback_Internal{},
+			Variant: &commonpb.Callback_Nexus_{
+				Nexus: &commonpb.Callback_Nexus{
+					Url: "http://localhost/cb",
+				},
 			},
 			Links: []*commonpb.Link{{
 				Variant: &commonpb.Link_BatchJob_{