Add task queue kind to NexusTask token (#11022)

## What

Add `TaskQueueKind` to the `NexusTask` token proto so the frontend can
pass the correct kind to matching on
`RespondNexusTaskCompleted`/`Failed`, instead of hardcoding
`TASK_QUEUE_KIND_NORMAL`.

## Why

To distinguish internal nexus invocations from user facing ones.
[#10141](https://github.com/temporalio/temporal/pull/10141) switched
`nexus_task_requests` `is_internal` detection from task queue name
prefix to `TaskQueueKind`, but the frontend hardcodes `NORMAL` when
constructing the matching request for Respond operations because the
token didn't carry the kind. This means `is_internal` is always `false`
for Respond on worker-commands queues.

## How did you test it?

Unit tests:
- Frontend handler: `RespondNexusTaskCompleted` and
`RespondNexusTaskFailed` preserve `TaskQueueKind` from token
(WORKER_COMMANDS preserved, NORMAL preserved, UNSPECIFIED defaults to
NORMAL)
- Matching engine: `PollNexusTaskQueue` sets the correct `TaskQueueKind`
in the serialized task token for both NORMAL and WORKER_COMMANDS
partitions
- Backwards compatible: old tokens without the field default to
`TASK_QUEUE_KIND_UNSPECIFIED`, treated as `NORMAL`

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

api/token/v1/message.pb.go

@@ -11,6 +11,7 @@ import (
 	sync "sync"
 	unsafe "unsafe"
 
+	v13 "go.temporal.io/api/enums/v1"
 	v12 "go.temporal.io/server/api/clock/v1"
 	v1 "go.temporal.io/server/api/history/v1"
 	v11 "go.temporal.io/server/api/persistence/v1"
@@ -448,6 +449,7 @@ type NexusTask struct {
 	NamespaceId   string                 `protobuf:"bytes,1,opt,name=namespace_id,json=namespaceId,proto3" json:"namespace_id,omitempty"`
 	TaskQueue     string                 `protobuf:"bytes,2,opt,name=task_queue,json=taskQueue,proto3" json:"task_queue,omitempty"`
 	TaskId        string                 `protobuf:"bytes,3,opt,name=task_id,json=taskId,proto3" json:"task_id,omitempty"`
+	TaskQueueKind v13.TaskQueueKind      `protobuf:"varint,4,opt,name=task_queue_kind,json=taskQueueKind,proto3,enum=temporal.api.enums.v1.TaskQueueKind" json:"task_queue_kind,omitempty"`
 	unknownFields protoimpl.UnknownFields
 	sizeCache     protoimpl.SizeCache
 }
@@ -503,6 +505,13 @@ func (x *NexusTask) GetTaskId() string {
 	return ""
 }
 
+func (x *NexusTask) GetTaskQueueKind() v13.TaskQueueKind {
+	if x != nil {
+		return x.TaskQueueKind
+	}
+	return v13.TaskQueueKind(0)
+}
+
 // A reference for loading a history event.
 type HistoryEventRef struct {
 	state protoimpl.MessageState `protogen:"open.v1"`
@@ -655,7 +664,7 @@ var File_temporal_server_api_token_v1_message_proto protoreflect.FileDescriptor
 
 const file_temporal_server_api_token_v1_message_proto_rawDesc = "" +
 	"\n" +
-	"*temporal/server/api/token/v1/message.proto\x12\x1ctemporal.server.api.token.v1\x1a\x1fgoogle/protobuf/timestamp.proto\x1a*temporal/server/api/clock/v1/message.proto\x1a,temporal/server/api/history/v1/message.proto\x1a,temporal/server/api/persistence/v1/hsm.proto\"\xd4\x03\n" +
+	"*temporal/server/api/token/v1/message.proto\x12\x1ctemporal.server.api.token.v1\x1a\x1fgoogle/protobuf/timestamp.proto\x1a&temporal/api/enums/v1/task_queue.proto\x1a*temporal/server/api/clock/v1/message.proto\x1a,temporal/server/api/history/v1/message.proto\x1a,temporal/server/api/persistence/v1/hsm.proto\"\xd4\x03\n" +
 	"\x13HistoryContinuation\x12\x15\n" +
 	"\x06run_id\x18\x01 \x01(\tR\x05runId\x12$\n" +
 	"\x0efirst_event_id\x18\x02 \x01(\x03R\ffirstEventId\x12\"\n" +
@@ -702,12 +711,13 @@ const file_temporal_server_api_token_v1_message_proto_rawDesc = "" +
 	"\fnamespace_id\x18\x01 \x01(\tR\vnamespaceId\x12\x1d\n" +
 	"\n" +
 	"task_queue\x18\x02 \x01(\tR\ttaskQueue\x12\x17\n" +
-	"\atask_id\x18\x03 \x01(\tR\x06taskId\"f\n" +
+	"\atask_id\x18\x03 \x01(\tR\x06taskId\"\xb4\x01\n" +
 	"\tNexusTask\x12!\n" +
 	"\fnamespace_id\x18\x01 \x01(\tR\vnamespaceId\x12\x1d\n" +
 	"\n" +
 	"task_queue\x18\x02 \x01(\tR\ttaskQueue\x12\x17\n" +
-	"\atask_id\x18\x03 \x01(\tR\x06taskId\"R\n" +
+	"\atask_id\x18\x03 \x01(\tR\x06taskId\x12L\n" +
+	"\x0ftask_queue_kind\x18\x04 \x01(\x0e2$.temporal.api.enums.v1.TaskQueueKindR\rtaskQueueKind\"R\n" +
 	"\x0fHistoryEventRef\x12\x19\n" +
 	"\bevent_id\x18\x01 \x01(\x03R\aeventId\x12$\n" +
 	"\x0eevent_batch_id\x18\x02 \x01(\x03R\feventBatchId\"\x80\x02\n" +
@@ -747,7 +757,8 @@ var file_temporal_server_api_token_v1_message_proto_goTypes = []any{
 	(*v1.VersionHistories)(nil),      // 9: temporal.server.api.history.v1.VersionHistories
 	(*v12.VectorClock)(nil),          // 10: temporal.server.api.clock.v1.VectorClock
 	(*timestamppb.Timestamp)(nil),    // 11: google.protobuf.Timestamp
-	(*v11.StateMachineRef)(nil),      // 12: temporal.server.api.persistence.v1.StateMachineRef
+	(v13.TaskQueueKind)(0),           // 12: temporal.api.enums.v1.TaskQueueKind
+	(*v11.StateMachineRef)(nil),      // 13: temporal.server.api.persistence.v1.StateMachineRef
 }
 var file_temporal_server_api_token_v1_message_proto_depIdxs = []int32{
 	7,  // 0: temporal.server.api.token.v1.HistoryContinuation.version_history_item:type_name -> temporal.server.api.history.v1.VersionHistoryItem
@@ -755,12 +766,13 @@ var file_temporal_server_api_token_v1_message_proto_depIdxs = []int32{
 	9,  // 2: temporal.server.api.token.v1.RawHistoryContinuation.version_histories:type_name -> temporal.server.api.history.v1.VersionHistories
 	10, // 3: temporal.server.api.token.v1.Task.clock:type_name -> temporal.server.api.clock.v1.VectorClock
 	11, // 4: temporal.server.api.token.v1.Task.started_time:type_name -> google.protobuf.Timestamp
-	12, // 5: temporal.server.api.token.v1.NexusOperationCompletion.ref:type_name -> temporal.server.api.persistence.v1.StateMachineRef
-	6,  // [6:6] is the sub-list for method output_type
-	6,  // [6:6] is the sub-list for method input_type
-	6,  // [6:6] is the sub-list for extension type_name
-	6,  // [6:6] is the sub-list for extension extendee
-	0,  // [0:6] is the sub-list for field type_name
+	12, // 5: temporal.server.api.token.v1.NexusTask.task_queue_kind:type_name -> temporal.api.enums.v1.TaskQueueKind
+	13, // 6: temporal.server.api.token.v1.NexusOperationCompletion.ref:type_name -> temporal.server.api.persistence.v1.StateMachineRef
+	7,  // [7:7] is the sub-list for method output_type
+	7,  // [7:7] is the sub-list for method input_type
+	7,  // [7:7] is the sub-list for extension type_name
+	7,  // [7:7] is the sub-list for extension extendee
+	0,  // [0:7] is the sub-list for field type_name
 }
 
 func init() { file_temporal_server_api_token_v1_message_proto_init() }

proto/internal/temporal/server/api/token/v1/message.proto

@@ -3,6 +3,7 @@ syntax = "proto3";
 package temporal.server.api.token.v1;
 
 import "google/protobuf/timestamp.proto";
+import "temporal/api/enums/v1/task_queue.proto";
 import "temporal/server/api/clock/v1/message.proto";
 import "temporal/server/api/history/v1/message.proto";
 import "temporal/server/api/persistence/v1/hsm.proto";
@@ -63,6 +64,7 @@ message NexusTask {
   string namespace_id = 1;
   string task_queue = 2;
   string task_id = 3;
+  temporal.api.enums.v1.TaskQueueKind task_queue_kind = 4;
 }
 
 // A reference for loading a history event.

service/frontend/workflow_handler.go

@@ -6253,11 +6253,15 @@ func (wh *WorkflowHandler) RespondNexusTaskCompleted(ctx context.Context, reques
 		}
 	}
 
+	taskQueueKind := tt.GetTaskQueueKind()
+	if taskQueueKind == enumspb.TASK_QUEUE_KIND_UNSPECIFIED {
+		taskQueueKind = enumspb.TASK_QUEUE_KIND_NORMAL
+	}
 	matchingRequest := &matchingservice.RespondNexusTaskCompletedRequest{
 		NamespaceId: namespaceId.String(),
 		TaskQueue: &taskqueuepb.TaskQueue{
 			Name: tt.GetTaskQueue(),
-			Kind: enumspb.TASK_QUEUE_KIND_NORMAL,
+			Kind: taskQueueKind,
 		},
 		TaskId:  tt.GetTaskId(),
 		Request: request,
@@ -6306,11 +6310,15 @@ func (wh *WorkflowHandler) RespondNexusTaskFailed(ctx context.Context, request *
 	// doesn't go into workflow history, and the Nexus request caller is unknown, there doesn't seem like there's a
 	// good reason to fail at this point.
 
+	taskQueueKind := tt.GetTaskQueueKind()
+	if taskQueueKind == enumspb.TASK_QUEUE_KIND_UNSPECIFIED {
+		taskQueueKind = enumspb.TASK_QUEUE_KIND_NORMAL
+	}
 	matchingRequest := &matchingservice.RespondNexusTaskFailedRequest{
 		NamespaceId: namespaceId.String(),
 		TaskQueue: &taskqueuepb.TaskQueue{
 			Name: tt.GetTaskQueue(),
-			Kind: enumspb.TASK_QUEUE_KIND_NORMAL,
+			Kind: taskQueueKind,
 		},
 		TaskId:  tt.GetTaskId(),
 		Request: request,

service/frontend/workflow_handler_test.go

@@ -37,6 +37,7 @@ import (
 	"go.temporal.io/server/api/matchingservicemock/v1"
 	persistencespb "go.temporal.io/server/api/persistence/v1"
 	taskqueuespb "go.temporal.io/server/api/taskqueue/v1"
+	tokenspb "go.temporal.io/server/api/token/v1"
 	"go.temporal.io/server/chasm/lib/callback"
 	"go.temporal.io/server/chasm/lib/nexusoperation"
 	"go.temporal.io/server/chasm/lib/workflow"
@@ -247,6 +248,116 @@ func (s *WorkflowHandlerSuite) getWorkflowHandler(config *Config) *WorkflowHandl
 	)
 }
 
+func (s *WorkflowHandlerSuite) TestRespondNexusTaskCompleted_PreservesTaskQueueKindFromToken() {
+	config := s.newConfig()
+	wh := s.getWorkflowHandler(config)
+
+	testCases := []struct {
+		name         string
+		kind         enumspb.TaskQueueKind
+		expectedKind enumspb.TaskQueueKind
+	}{
+		{
+			name:         "worker commands kind preserved",
+			kind:         enumspb.TASK_QUEUE_KIND_WORKER_COMMANDS,
+			expectedKind: enumspb.TASK_QUEUE_KIND_WORKER_COMMANDS,
+		},
+		{
+			name:         "normal kind preserved",
+			kind:         enumspb.TASK_QUEUE_KIND_NORMAL,
+			expectedKind: enumspb.TASK_QUEUE_KIND_NORMAL,
+		},
+		{
+			name:         "unspecified defaults to normal",
+			kind:         enumspb.TASK_QUEUE_KIND_UNSPECIFIED,
+			expectedKind: enumspb.TASK_QUEUE_KIND_NORMAL,
+		},
+	}
+
+	for _, tc := range testCases {
+		s.Run(tc.name, func() {
+			token, err := s.tokenSerializer.SerializeNexusTaskToken(&tokenspb.NexusTask{
+				NamespaceId:   s.testNamespaceID.String(),
+				TaskQueue:     "test-tq",
+				TaskId:        "test-task-id",
+				TaskQueueKind: tc.kind,
+			})
+			s.NoError(err)
+
+			s.mockMatchingClient.EXPECT().
+				RespondNexusTaskCompleted(gomock.Any(), gomock.Any(), gomock.Any()).
+				DoAndReturn(func(_ context.Context, req *matchingservice.RespondNexusTaskCompletedRequest, _ ...grpc.CallOption) (*matchingservice.RespondNexusTaskCompletedResponse, error) {
+					s.Equal(tc.expectedKind, req.GetTaskQueue().GetKind())
+					return &matchingservice.RespondNexusTaskCompletedResponse{}, nil
+				})
+
+			_, err = wh.RespondNexusTaskCompleted(context.Background(), &workflowservice.RespondNexusTaskCompletedRequest{
+				Namespace: s.testNamespace.String(),
+				TaskToken: token,
+			})
+			s.NoError(err)
+		})
+	}
+}
+
+func (s *WorkflowHandlerSuite) TestRespondNexusTaskFailed_PreservesTaskQueueKindFromToken() {
+	config := s.newConfig()
+	wh := s.getWorkflowHandler(config)
+
+	testCases := []struct {
+		name         string
+		kind         enumspb.TaskQueueKind
+		expectedKind enumspb.TaskQueueKind
+	}{
+		{
+			name:         "worker commands kind preserved",
+			kind:         enumspb.TASK_QUEUE_KIND_WORKER_COMMANDS,
+			expectedKind: enumspb.TASK_QUEUE_KIND_WORKER_COMMANDS,
+		},
+		{
+			name:         "normal kind preserved",
+			kind:         enumspb.TASK_QUEUE_KIND_NORMAL,
+			expectedKind: enumspb.TASK_QUEUE_KIND_NORMAL,
+		},
+		{
+			name:         "unspecified defaults to normal",
+			kind:         enumspb.TASK_QUEUE_KIND_UNSPECIFIED,
+			expectedKind: enumspb.TASK_QUEUE_KIND_NORMAL,
+		},
+	}
+
+	for _, tc := range testCases {
+		s.Run(tc.name, func() {
+			token, err := s.tokenSerializer.SerializeNexusTaskToken(&tokenspb.NexusTask{
+				NamespaceId:   s.testNamespaceID.String(),
+				TaskQueue:     "test-tq",
+				TaskId:        "test-task-id",
+				TaskQueueKind: tc.kind,
+			})
+			s.NoError(err)
+
+			s.mockMatchingClient.EXPECT().
+				RespondNexusTaskFailed(gomock.Any(), gomock.Any(), gomock.Any()).
+				DoAndReturn(func(_ context.Context, req *matchingservice.RespondNexusTaskFailedRequest, _ ...grpc.CallOption) (*matchingservice.RespondNexusTaskFailedResponse, error) {
+					s.Equal(tc.expectedKind, req.GetTaskQueue().GetKind())
+					return &matchingservice.RespondNexusTaskFailedResponse{}, nil
+				})
+
+			_, err = wh.RespondNexusTaskFailed(context.Background(), &workflowservice.RespondNexusTaskFailedRequest{
+				Namespace: s.testNamespace.String(),
+				TaskToken: token,
+				Failure: &failurepb.Failure{
+					Message: "test failure",
+					FailureInfo: &failurepb.Failure_NexusHandlerFailureInfo{
+						NexusHandlerFailureInfo: &failurepb.NexusHandlerFailureInfo{},
+					},
+				},
+			})
+			s.NoError(err)
+		})
+	}
+}
+
 func (s *WorkflowHandlerSuite) TestCheckWorkerDeploymentReadRateLimitResourceExhaustedScope() {
 	rateLimiter := quotas.NewMockRequestRateLimiter(s.controller)
 	wh := &WorkflowHandler{

service/matching/matching_engine.go

@@ -2766,9 +2766,10 @@ pollLoop:
 		}
 
 		taskToken := &tokenspb.NexusTask{
-			NamespaceId: string(namespaceID),
-			TaskQueue:   taskQueueName,
-			TaskId:      task.nexus.taskID,
+			NamespaceId:   string(namespaceID),
+			TaskQueue:     taskQueueName,
+			TaskId:        task.nexus.taskID,
+			TaskQueueKind: partition.Kind(),
 		}
 		serializedToken, _ := e.tokenSerializer.SerializeNexusTaskToken(taskToken)
 

service/matching/matching_engine_test.go

@@ -4144,6 +4144,93 @@ func (s *matchingEngineSuite) TestDispatchNexusTask_ValidateTimeoutBuffer() {
 	}
 }
 
+func (s *matchingEngineSuite) TestPollNexusTaskQueue_TaskTokenContainsTaskQueueKind() {
+	testCases := []struct {
+		name         string
+		kind         enumspb.TaskQueueKind
+		expectedKind enumspb.TaskQueueKind
+	}{
+		{
+			name:         "normal kind",
+			kind:         enumspb.TASK_QUEUE_KIND_NORMAL,
+			expectedKind: enumspb.TASK_QUEUE_KIND_NORMAL,
+		},
+		{
+			name:         "worker commands kind",
+			kind:         enumspb.TASK_QUEUE_KIND_WORKER_COMMANDS,
+			expectedKind: enumspb.TASK_QUEUE_KIND_WORKER_COMMANDS,
+		},
+	}
+
+	for _, tc := range testCases {
+		s.Run(tc.name, func() {
+			namespaceID := s.ns.ID().String()
+			taskQueueName := "test-nexus-tq"
+
+			dispatchReq := &matchingservice.DispatchNexusTaskRequest{
+				NamespaceId: namespaceID,
+				TaskQueue: &taskqueuepb.TaskQueue{
+					Name: taskQueueName,
+					Kind: tc.kind,
+				},
+				Request: &nexuspb.Request{
+					Header: map[string]string{
+						"request-timeout": "10s",
+					},
+				},
+			}
+
+			nexusTask := newInternalNexusTask(
+				"test-task-id",
+				time.Now().Add(10*time.Second),
+				time.Time{},
+				dispatchReq,
+			)
+
+			partition, err := tqid.PartitionFromProto(
+				&taskqueuepb.TaskQueue{Name: taskQueueName, Kind: tc.kind},
+				namespaceID,
+				enumspb.TASK_QUEUE_TYPE_NEXUS,
+			)
+			s.Require().NoError(err)
+
+			mockPM := NewMocktaskQueuePartitionManager(s.controller)
+			mockPM.EXPECT().WaitUntilInitialized(gomock.Any()).Return(nil)
+			mockPM.EXPECT().LongPollExpirationInterval().Return(time.Minute)
+			mockPM.EXPECT().Stop(gomock.Any()).AnyTimes()
+			mockPM.EXPECT().PollTask(gomock.Any(), gomock.Any()).Return(nexusTask, false, nil)
+
+			s.matchingEngine.partitionsLock.Lock()
+			s.matchingEngine.partitions[partition.Key()] = mockPM
+			s.matchingEngine.partitionsLock.Unlock()
+			s.matchingEngine.nexusResults = collection.NewSyncMap[string, chan *nexusResult]()
+			s.matchingEngine.outstandingPollers = collection.NewSyncMap[string, context.CancelFunc]()
+			s.matchingEngine.shutdownWorkers = cache.New(100, &cache.Options{TTL: 30 * time.Second})
+
+			resp, err := s.matchingEngine.PollNexusTaskQueue(
+				context.Background(),
+				&matchingservice.PollNexusTaskQueueRequest{
+					NamespaceId: namespaceID,
+					PollerId:    uuid.NewString(),
+					Request: &workflowservice.PollNexusTaskQueueRequest{
+						Namespace: string(s.ns.Name()),
+						TaskQueue: &taskqueuepb.TaskQueue{
+							Name: taskQueueName,
+							Kind: tc.kind,
+						},
+					},
+				},
+				metrics.NoopMetricsHandler,
+			)
+			s.Require().NoError(err)
+
+			token, err := s.matchingEngine.tokenSerializer.DeserializeNexusTaskToken(resp.GetResponse().GetTaskToken())
+			s.Require().NoError(err)
+			s.Require().Equal(tc.expectedKind, token.GetTaskQueueKind())
+		})
+	}
+}
+
 // Following are tests for SyncDeploymentUserData API when it uses the new deployment data format.
 
 // TestSyncDeploymentUserData_NewDeploymentDataRemovesOldVersions verifies that when a new routing config is set for a deployment using the latest deployment data format,