[SDK Ergonomics] NEXUS-519: Support Query-backed Nexus Operations

common/links/validator.go

@@ -57,6 +57,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,14 +52,24 @@ 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{
 			validWorkflowEvent,
 			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/query_workflow_test.go

@@ -140,6 +140,61 @@ func (s *QueryWorkflowSuite) TestQueryWorkflow_Consistent_PiggybackQuery() {
 	s.Equal("pauseabc", queryResultStr)
 }
 
+func (s *QueryWorkflowSuite) TestQueryWorkflowResult_ContainsWorkflowLink() {
+	env := testcore.NewEnv(s.T())
+	workflowFn := func(ctx workflow.Context) error {
+		_ = workflow.SetQueryHandler(ctx, "test", func() (string, error) {
+			return "", nil
+		})
+		workflow.GetSignalChannel(ctx, "test").Receive(ctx, nil)
+		return nil
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+
+	env.SdkWorker().RegisterWorkflow(workflowFn)
+
+	wid := "test-query-result-link-wid"
+	run, err := env.SdkClient().ExecuteWorkflow(ctx,
+		sdkclient.StartWorkflowOptions{ID: wid, TaskQueue: env.WorkerTaskQueue()}, workflowFn)
+	s.NoError(err)
+
+	// use frontend client to query workflow so that the resp can be
+	// inspected directly instead of using the sdkclient
+	resp, err := env.FrontendClient().QueryWorkflow(ctx, &workflowservice.QueryWorkflowRequest{
+		Namespace:            env.Namespace().String(),
+		Execution:            &commonpb.WorkflowExecution{WorkflowId: wid},
+		Query:                &querypb.WorkflowQuery{QueryType: "test"},
+		QueryRejectCondition: enumspb.QUERY_REJECT_CONDITION_NOT_OPEN, // set here too so the second query below differs only by the workflow being closed
+	})
+	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("Query processed", link.GetReason())
+
+	s.NoError(env.SdkClient().SignalWorkflow(ctx, wid, "", "test", ""))
+	s.NoError(run.Get(ctx, nil))
+	// run same after workflow closed, should reject due to QueryRejectCondition
+	resp, err = env.FrontendClient().QueryWorkflow(ctx, &workflowservice.QueryWorkflowRequest{
+		Namespace:            env.Namespace().String(),
+		Execution:            &commonpb.WorkflowExecution{WorkflowId: wid},
+		Query:                &querypb.WorkflowQuery{QueryType: "test"},
+		QueryRejectCondition: enumspb.QUERY_REJECT_CONDITION_NOT_OPEN,
+	})
+	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("Query rejected", link.GetReason())
+}
+
 func (s *QueryWorkflowSuite) TestQueryWorkflow_QueryWhileBackoff() {
 	env := testcore.NewEnv(s.T())
 	tv := testvars.New(s.T())