[SDK Ergonomics] NEXUS-519: Support Query-backed Nexus Operations (#11274)

## What changed?
Adds server support for WorkflowQuery-backed Nexus Operations 

## Why?
Part of effort to expose all Temporal primitives as Nexus Operations 

## How did you test it?
- [x] built
- [ ] run locally and tested manually
- [ ] covered by existing tests
- [ ] added new unit test(s)
- [x] added new functional test(s)

## TODO:
- [x] https://github.com/temporalio/api/pull/842

common/links/validator.go

@@ -72,6 +72,16 @@ func Validate(links []*commonpb.Link, maxAllowedLinks, maxSize int) error {
 			if t.Activity.GetRunId() == "" {
 				return serviceerror.NewInvalidArgument("activity link must not have an empty run ID field")
 			}
+		case *commonpb.Link_Workflow_:
+			if t.Workflow.GetNamespace() == "" {
+				return serviceerror.NewInvalidArgument("workflow link must not have an empty namespace field")
+			}
+			if t.Workflow.GetWorkflowId() == "" {
+				return serviceerror.NewInvalidArgument("workflow link must not have an empty workflow ID field")
+			}
+			if t.Workflow.GetRunId() == "" {
+				return serviceerror.NewInvalidArgument("workflow link must not have an empty run ID field")
+			}
 		default:
 			return serviceerror.NewInvalidArgument("unsupported link variant")
 		}

common/links/validator_test.go

@@ -52,6 +52,15 @@ func TestValidate(t *testing.T) {
 			},
 		},
 	}
+	validWorkflow := &commonpb.Link{
+		Variant: &commonpb.Link_Workflow_{
+			Workflow: &commonpb.Link_Workflow{
+				Namespace:  "ns",
+				WorkflowId: "wid",
+				RunId:      "rid",
+			},
+		},
+	}
 
 	t.Run("HappyPath", func(t *testing.T) {
 		err := links.Validate([]*commonpb.Link{
@@ -59,7 +68,8 @@ func TestValidate(t *testing.T) {
 			validBatchJob,
 			validNexusOperation,
 			validActivity,
-		}, maxLinks+1, maxSize)
+			validWorkflow,
+		}, maxLinks+2, maxSize)
 		require.NoError(t, err)
 	})
 
@@ -155,4 +165,25 @@ func TestValidate(t *testing.T) {
 		err := links.Validate([]*commonpb.Link{{}}, maxLinks, maxSize)
 		require.ErrorContains(t, err, "unsupported link variant")
 	})
+
+	t.Run("Workflow/EmptyNamespace", func(t *testing.T) {
+		l := proto.Clone(validWorkflow).(*commonpb.Link)
+		l.GetWorkflow().Namespace = ""
+		err := links.Validate([]*commonpb.Link{l}, maxLinks, maxSize)
+		require.EqualError(t, err, "workflow link must not have an empty namespace field")
+	})
+
+	t.Run("Workflow/EmptyWorkflowID", func(t *testing.T) {
+		l := proto.Clone(validWorkflow).(*commonpb.Link)
+		l.GetWorkflow().WorkflowId = ""
+		err := links.Validate([]*commonpb.Link{l}, maxLinks, maxSize)
+		require.EqualError(t, err, "workflow link must not have an empty workflow ID field")
+	})
+
+	t.Run("Workflow/EmptyRunID", func(t *testing.T) {
+		l := proto.Clone(validWorkflow).(*commonpb.Link)
+		l.GetWorkflow().RunId = ""
+		err := links.Validate([]*commonpb.Link{l}, maxLinks, maxSize)
+		require.EqualError(t, err, "workflow link must not have an empty run ID field")
+	})
 }

common/nexus/links.go

@@ -5,13 +5,14 @@ import (
 
 	"github.com/nexus-rpc/sdk-go/nexus"
 	commonpb "go.temporal.io/api/common/v1"
+	"go.temporal.io/api/temporalnexus"
 	"go.temporal.io/server/common/log"
 	"go.temporal.io/server/common/log/tag"
 )
 
 // ConvertNexusLinksToProtoLinks converts a slice of Nexus SDK links into Temporal proto links,
-// supporting Link_WorkflowEvent and Link_Activity variants. Unsupported or malformed entries
-// are skipped with a warning since links are non-essential to execution.
+// supporting Link_Workflow, Link_WorkflowEvent, and Link_Activity variants. Unsupported or
+// malformed entries are skipped with a warning since links are non-essential to execution.
 func ConvertNexusLinksToProtoLinks(nexusLinks []nexus.Link, logger log.Logger) []*commonpb.Link {
 	var out []*commonpb.Link
 	for _, nexusLink := range nexusLinks {
@@ -40,6 +41,18 @@ func ConvertNexusLinksToProtoLinks(nexusLinks []nexus.Link, logger log.Logger) [
 			out = append(out, &commonpb.Link{
 				Variant: &commonpb.Link_Activity_{Activity: link},
 			})
+		case string((&commonpb.Link_Workflow{}).ProtoReflect().Descriptor().FullName()):
+			link, err := temporalnexus.ConvertNexusLinkToLinkWorkflow(nexusLink)
+			if err != nil {
+				logger.Warn(
+					fmt.Sprintf("failed to parse link to %q: %s", nexusLink.Type, nexusLink.URL),
+					tag.Error(err),
+				)
+				continue
+			}
+			out = append(out, &commonpb.Link{
+				Variant: &commonpb.Link_Workflow_{Workflow: link},
+			})
 		default:
 			logger.Warn(fmt.Sprintf("invalid link data type: %q", nexusLink.Type))
 		}

common/nexus/links_test.go

@@ -13,12 +13,10 @@ import (
 	"go.temporal.io/server/common/testing/protorequire"
 )
 
-// TestConvertNexusLinksToProtoLinks_ActivityVariant verifies that the shared
-// converter handles both WorkflowEvent and Activity link variants, drops
-// unsupported types, and skips malformed entries — exercised by the Nexus task
-// handler's start-response flow so a SAA invoked from a Nexus operation can
-// surface its Activity link back to the caller.
-func TestConvertNexusLinksToProtoLinks_ActivityVariant(t *testing.T) {
+// TestConvertNexusLinksToProtoLinks verifies that the converter handles the
+// Workflow, WorkflowEvent, and Activity link variants, drops unsupported types,
+// and skips malformed entries.
+func TestConvertNexusLinksToProtoLinks(t *testing.T) {
 	logger := log.NewTestLogger()
 
 	workflowEvent := nexus.Link{
@@ -36,6 +34,13 @@ func TestConvertNexusLinksToProtoLinks_ActivityVariant(t *testing.T) {
 		},
 		Type: "temporal.api.common.v1.Link.Activity",
 	}
+	workflow := nexus.Link{
+		URL: &url.URL{
+			Scheme: "temporal",
+			Path:   "/namespaces/ns/workflows/wf-id/run-id",
+		},
+		Type: "temporal.api.common.v1.Link.Workflow",
+	}
 	unsupported := nexus.Link{
 		URL:  &url.URL{Scheme: "temporal", Path: "/foo"},
 		Type: "unknown.Type",
@@ -44,9 +49,29 @@ func TestConvertNexusLinksToProtoLinks_ActivityVariant(t *testing.T) {
 		URL:  &url.URL{Scheme: "temporal", Path: "/namespaces/ns/foo/act-id"},
 		Type: "temporal.api.common.v1.Link.Activity",
 	}
+	malformedWorkflows := []nexus.Link{
+		{
+			URL:  &url.URL{Scheme: "temporal", Path: "/namespaces//workflows/wid/rid"}, // missing ns
+			Type: "temporal.api.common.v1.Link.Workflow",
+		},
+		{
+			URL:  &url.URL{Scheme: "temporal", Path: "/namespaces/ns/workflows//rid"}, // missing wid
+			Type: "temporal.api.common.v1.Link.Workflow",
+		},
+		{
+			URL:  &url.URL{Scheme: "temporal", Path: "/namespaces/ns/workflows/wid/"}, // missing rid
+			Type: "temporal.api.common.v1.Link.Workflow",
+		},
+		{
+			URL:  &url.URL{Scheme: "temporal", Path: "/foo"}, // incorrect path
+			Type: "temporal.api.common.v1.Link.Workflow",
+		},
+	}
+	nexusLinks := []nexus.Link{workflowEvent, activity, workflow, unsupported, malformedActivity}
+	nexusLinks = append(nexusLinks, malformedWorkflows...)
 
-	out := commonnexus.ConvertNexusLinksToProtoLinks([]nexus.Link{workflowEvent, activity, unsupported, malformedActivity}, logger)
-	require.Len(t, out, 2, "workflow-event and activity links must round-trip; unsupported and malformed entries must be dropped")
+	out := commonnexus.ConvertNexusLinksToProtoLinks(nexusLinks, logger)
+	require.Len(t, out, 3, "workflow, workflow-event, and activity links must round-trip; unsupported and malformed entries must be dropped")
 
 	expected := []*commonpb.Link{
 		{
@@ -73,6 +98,15 @@ func TestConvertNexusLinksToProtoLinks_ActivityVariant(t *testing.T) {
 				},
 			},
 		},
+		{
+			Variant: &commonpb.Link_Workflow_{
+				Workflow: &commonpb.Link_Workflow{
+					Namespace:  "ns",
+					WorkflowId: "wf-id",
+					RunId:      "run-id",
+				},
+			},
+		},
 	}
 	protorequire.ProtoSliceEqual(t, expected, out)
 }

service/history/api/queryworkflow/api.go

@@ -37,7 +37,7 @@ func Invoke(
 	workflowConsistencyChecker api.WorkflowConsistencyChecker,
 	rawMatchingClient matchingservice.MatchingServiceClient,
 	matchingClient matchingservice.MatchingServiceClient,
-) (_ *historyservice.QueryWorkflowResponse, retError error) {
+) (resp *historyservice.QueryWorkflowResponse, retError error) {
 	scope := shardContext.GetMetricsHandler().WithTags(metrics.OperationTag(metrics.HistoryQueryWorkflowScope))
 	namespaceID := namespace.ID(request.GetNamespaceId())
 	err := api.ValidateNamespaceUUID(namespaceID)
@@ -79,6 +79,27 @@ func Invoke(
 		// Note: QueryWorkflow should not alter mutable state, so it is safe to ignore error and not clear ms.
 		workflowLease.GetReleaseFn()(nil)
 	}()
+	defer func() {
+		if retError != nil || resp.GetResponse() == nil {
+			return
+		}
+		// Add link to Workflow regardless of query status. A rejection on the query is not an RPC error,
+		// so it gets the same link that a processed query would get - only the reason differs.
+		reason := "Query processed"
+		if resp.GetResponse().GetQueryRejected() != nil {
+			reason = "Query rejected"
+		}
+		resp.Response.Link = &commonpb.Link{
+			Variant: &commonpb.Link_Workflow_{
+				Workflow: &commonpb.Link_Workflow{
+					Namespace:  nsEntry.Name().String(),
+					WorkflowId: workflowKey.WorkflowID,
+					RunId:      workflowKey.RunID,
+					Reason:     reason,
+				},
+			},
+		}
+	}()
 
 	// Context metadata is automatically set during mutable state transaction close for operations that mutate state.
 	// Since QueryWorkflow is readonly and never closes the transaction, we explicitly call SetContextMetadata

tests/nexus_workflow_query_test.go

@@ -0,0 +1,105 @@
+package tests
+
+import (
+	"context"
+
+	"github.com/nexus-rpc/sdk-go/nexus"
+	commonpb "go.temporal.io/api/common/v1"
+	enumspb "go.temporal.io/api/enums/v1"
+	querypb "go.temporal.io/api/query/v1"
+	apitemporalnexus "go.temporal.io/api/temporalnexus"
+	"go.temporal.io/api/workflowservice/v1"
+	"go.temporal.io/sdk/client"
+	"go.temporal.io/sdk/worker"
+	"go.temporal.io/sdk/workflow"
+	"go.temporal.io/server/common/nexus/nexustest"
+	"go.temporal.io/server/common/payloads"
+	"go.temporal.io/server/tests/testcore"
+)
+
+func (s *NexusWorkflowTestSuite) TestNexusOperationBackedByQuery(chasmEnabled bool) {
+	env := s.newTestEnv(chasmEnabled)
+	ctx := s.Context()
+	taskQueue := testcore.RandomizeStr(s.T().Name())
+
+	queryType := "status"
+	signalName := "done"
+	handlerWorkflowID := testcore.RandomizeStr(s.T().Name() + "-handler")
+	handlerWf := func(ctx workflow.Context) error {
+		_ = workflow.SetQueryHandler(ctx, queryType, func() (string, error) {
+			return "handler-status", nil
+		})
+		workflow.GetSignalChannel(ctx, signalName).Receive(ctx, nil)
+		return nil
+	}
+
+	h := nexustest.Handler{
+		OnStartOperation: func(
+			ctx context.Context,
+			service, operation string,
+			input *nexus.LazyValue,
+			options nexus.StartOperationOptions,
+		) (nexus.HandlerStartOperationResult[any], error) {
+			resp, err := env.FrontendClient().QueryWorkflow(ctx, &workflowservice.QueryWorkflowRequest{
+				Namespace: env.Namespace().String(),
+				Execution: &commonpb.WorkflowExecution{WorkflowId: handlerWorkflowID},
+				Query:     &querypb.WorkflowQuery{QueryType: queryType},
+			})
+			if err != nil {
+				return nil, err
+			}
+			// simulate the stitching that will be done by the SDKs eventually
+			nexus.AddHandlerLinks(ctx, apitemporalnexus.ConvertLinkWorkflowToNexusLink(resp.GetLink().GetWorkflow()))
+
+			var result string
+			if err := payloads.Decode(resp.GetQueryResult(), &result); err != nil {
+				return nil, err
+			}
+			return &nexus.HandlerStartOperationResultSync[any]{Value: result}, nil
+		},
+	}
+	endpointName := env.createRandomExternalNexusServer(ctx, s.T(), h)
+
+	callerWF := func(ctx workflow.Context) (string, error) {
+		c := workflow.NewNexusClient(endpointName, "service")
+		fut := c.ExecuteOperation(ctx, "operation", "input", workflow.NexusOperationOptions{})
+		var result string
+		err := fut.Get(ctx, &result)
+		return result, err
+	}
+
+	w := worker.New(env.SdkClient(), taskQueue, worker.Options{})
+	w.RegisterWorkflow(callerWF)
+	w.RegisterWorkflow(handlerWf)
+	s.NoError(w.Start())
+	defer w.Stop()
+
+	handlerRun, err := env.SdkClient().ExecuteWorkflow(ctx, client.StartWorkflowOptions{
+		ID:        handlerWorkflowID,
+		TaskQueue: taskQueue,
+	}, handlerWf)
+	s.NoError(err)
+
+	callerRun, err := env.SdkClient().ExecuteWorkflow(ctx, client.StartWorkflowOptions{
+		TaskQueue: taskQueue,
+	}, callerWF)
+	s.NoError(err)
+
+	var result string
+	s.NoError(callerRun.Get(ctx, &result))
+	s.Equal("handler-status", result)
+
+	// verify the nexus operation completed event carries a link to the handler's workflow
+	hist := env.GetHistory(env.Namespace().String(), &commonpb.WorkflowExecution{WorkflowId: callerRun.GetID()})
+	completedEvent := s.RequireHistoryEvent(hist, enumspb.EVENT_TYPE_NEXUS_OPERATION_COMPLETED)
+	s.Len(completedEvent.GetLinks(), 1)
+	workflowLink := completedEvent.GetLinks()[0].GetWorkflow()
+	s.NotNil(workflowLink, "completed event must carry a link of type workflow")
+	s.Equal(env.Namespace().String(), workflowLink.GetNamespace())
+	s.Equal(handlerWorkflowID, workflowLink.GetWorkflowId())
+	s.Equal(handlerRun.GetRunID(), workflowLink.GetRunId())
+	s.Equal("Query processed", workflowLink.GetReason())
+
+	s.NoError(env.SdkClient().SignalWorkflow(ctx, handlerWorkflowID, "", signalName, nil))
+	s.NoError(handlerRun.Get(ctx, nil))
+}

tests/query_workflow_test.go

@@ -21,6 +21,7 @@ import (
 	"go.temporal.io/api/workflowservice/v1"
 	"go.temporal.io/sdk/client"
 	sdkclient "go.temporal.io/sdk/client"
+	"go.temporal.io/sdk/converter"
 	"go.temporal.io/sdk/worker"
 	"go.temporal.io/sdk/workflow"
 	"go.temporal.io/server/common/dynamicconfig"
@@ -140,6 +141,130 @@ func (s *QueryWorkflowSuite) TestQueryWorkflow_Consistent_PiggybackQuery() {
 	s.Equal("pauseabc", queryResultStr)
 }
 
+func (s *QueryWorkflowSuite) TestQueryWorkflowResult_ContainsWorkflowLink() {
+	env := testcore.NewEnv(s.T())
+	queryName := "query"
+	signalName := "test"
+	workflowFn := func(ctx workflow.Context) error {
+		orderStatus := "initialized"
+		_ = workflow.SetQueryHandler(ctx, queryName, func(_ string) (string, error) {
+			return orderStatus, nil
+		})
+		workflow.GetSignalChannel(ctx, signalName).Receive(ctx, &orderStatus)
+		return nil
+	}
+	ctx := s.Context()
+
+	env.SdkWorker().RegisterWorkflow(workflowFn)
+
+	wid := "nexus-query-workflow-tests"
+	run, err := env.SdkClient().ExecuteWorkflow(
+		ctx,
+		sdkclient.StartWorkflowOptions{
+			ID:        wid,
+			TaskQueue: env.WorkerTaskQueue()},
+		workflowFn,
+	)
+	s.NoError(err)
+
+	type testCase struct {
+		name                 string
+		query                *querypb.WorkflowQuery
+		err                  bool
+		reason               string
+		result               string
+		queryRejectCondition enumspb.QueryRejectCondition
+	}
+
+	runTestCase := func(tc testCase) {
+		resp, err := env.FrontendClient().QueryWorkflow(ctx, &workflowservice.QueryWorkflowRequest{
+			Namespace:            env.Namespace().String(),
+			Execution:            &commonpb.WorkflowExecution{WorkflowId: wid},
+			Query:                tc.query,
+			QueryRejectCondition: tc.queryRejectCondition,
+		})
+		if tc.err {
+			s.Error(err)
+			return
+		}
+		s.NoError(err)
+		link := resp.GetLink().GetWorkflow()
+		s.NotNil(link, "query must carry a link of type workflow")
+		s.Equal(env.Namespace().String(), link.GetNamespace())
+		s.Equal(wid, link.GetWorkflowId())
+		s.Equal(run.GetRunID(), link.GetRunId())
+		s.Equal(tc.reason, link.GetReason())
+		if tc.result != "" {
+			s.Equal(tc.result, testcore.DecodeString(s.T(), resp.QueryResult))
+		}
+	}
+
+	runCases := func(groupName string, cases []testCase) {
+		for _, tc := range cases {
+			s.T().Run(groupName+"/"+tc.name, func(t *testing.T) { //nolint:testifylint // subtests need serialized execution(running, completed)
+				runTestCase(tc)
+			})
+		}
+	}
+
+	// TCs that are expected to fail on both running and completed workflows identically
+	invalidTestCases := []testCase{
+		{
+			name: "malformed args",
+			query: &querypb.WorkflowQuery{
+				QueryType: queryName,
+				QueryArgs: &commonpb.Payloads{
+					Payloads: []*commonpb.Payload{{
+						Metadata: map[string][]byte{
+							converter.MetadataEncoding: []byte(converter.MetadataEncodingJSON),
+						},
+						Data: []byte("dummy data"), // invalid JSON, fails to decode
+					}},
+				},
+			},
+			err: true,
+		},
+		{
+			name:  "unknown query type",
+			query: &querypb.WorkflowQuery{QueryType: "some unknown query"},
+			err:   true,
+		},
+	}
+
+	runningWorkflowTestCases := []testCase{
+		{
+			name:   "well-formed query",
+			query:  &querypb.WorkflowQuery{QueryType: queryName},
+			reason: "Query processed",
+			result: "initialized",
+		},
+	}
+	runCases("running-workflow", runningWorkflowTestCases)
+	runCases("running-workflow", invalidTestCases)
+
+	// set the query result via signal so that query "reload" on completed workflow is verified inline
+	s.NoError(env.SdkClient().SignalWorkflow(ctx, wid, "", signalName, "order-delivered"))
+	s.NoError(run.Get(ctx, nil))
+
+	completedWorkflowTestCases := []testCase{
+		{
+			name:   "well-formed replayable query succeeds",
+			query:  &querypb.WorkflowQuery{QueryType: queryName},
+			reason: "Query processed",
+			result: "order-delivered",
+		},
+		{
+			name:                 "well-formed non-replayable query fails",
+			query:                &querypb.WorkflowQuery{QueryType: queryName},
+			reason:               "Query rejected",
+			queryRejectCondition: enumspb.QUERY_REJECT_CONDITION_NOT_OPEN,
+		},
+	}
+
+	runCases("completed-workflow", completedWorkflowTestCases)
+	runCases("completed-workflow", invalidTestCases)
+}
+
 func (s *QueryWorkflowSuite) TestQueryWorkflow_QueryWhileBackoff() {
 	env := testcore.NewEnv(s.T())
 	tv := testvars.New(s.T())