First pass at adding links for signalling

Author: Evanthx

temporal-sdk/src/main/java/io/temporal/internal/client/RootWorkflowClientInvoker.java

@@ -120,6 +120,12 @@ public WorkflowSignalOutput signal(WorkflowSignalInput input) {
             .setRequestId(UUID.randomUUID().toString())
             .setHeader(HeaderUtils.toHeaderGrpc(input.getHeader(), null));
 
+    // If this signal is being issued from inside a Nexus operation handler, forward the inbound
+    // Nexus task links so the SignalWorkflowExecution history event links back to the caller.
+    if (CurrentNexusOperationContext.isNexusContext()) {
+      request.addAllLinks(CurrentNexusOperationContext.get().getNexusOperationLinks());
+    }
+
     DataConverter dataConverterWitSignalContext =
         clientOptions
             .getDataConverter()
@@ -129,7 +135,12 @@ public WorkflowSignalOutput signal(WorkflowSignalInput input) {
 
     Optional<Payloads> inputArgs = dataConverterWitSignalContext.toPayloads(input.getArguments());
     inputArgs.ifPresent(request::setInput);
-    genericClient.signal(request.build());
+    SignalWorkflowExecutionResponse response = genericClient.signal(request.build());
+    // Server >=1.31 with EnableCHASMSignalBacklinks returns a backlink pointing at the signal
+    // event; older servers leave it unset. Propagate when present.
+    if (CurrentNexusOperationContext.isNexusContext() && response.hasLink()) {
+      CurrentNexusOperationContext.get().setSignalWorkflowResponseLink(response.getLink());
+    }
     return new WorkflowSignalOutput();
   }
 
@@ -148,17 +159,27 @@ public WorkflowSignalWithStartOutput signalWithStart(WorkflowSignalWithStartInpu
 
     Optional<Payloads> signalInput =
         dataConverterWithWorkflowContext.toPayloads(input.getSignalArguments());
-    SignalWithStartWorkflowExecutionRequest request =
-        requestsHelper
-            .newSignalWithStartWorkflowExecutionRequest(
-                startRequest, input.getSignalName(), signalInput.orElse(null))
-            .build();
+    SignalWithStartWorkflowExecutionRequest.Builder requestBuilder =
+        requestsHelper.newSignalWithStartWorkflowExecutionRequest(
+            startRequest, input.getSignalName(), signalInput.orElse(null));
+    // If this signalWithStart is being issued from inside a Nexus operation handler, forward
+    // the inbound Nexus task links so both the WorkflowExecutionStarted and
+    // WorkflowExecutionSignaled events on the callee link back to the caller.
+    if (CurrentNexusOperationContext.isNexusContext()) {
+      requestBuilder.addAllLinks(CurrentNexusOperationContext.get().getNexusOperationLinks());
+    }
+    SignalWithStartWorkflowExecutionRequest request = requestBuilder.build();
     SignalWithStartWorkflowExecutionResponse response = genericClient.signalWithStart(request);
     WorkflowExecution execution =
         WorkflowExecution.newBuilder()
             .setRunId(response.getRunId())
             .setWorkflowId(request.getWorkflowId())
             .build();
+    // Server >=1.31 with EnableCHASMSignalBacklinks returns a backlink pointing at the signal
+    // event; older servers leave it unset. Propagate when present.
+    if (CurrentNexusOperationContext.isNexusContext() && response.hasSignalLink()) {
+      CurrentNexusOperationContext.get().setSignalWorkflowResponseLink(response.getSignalLink());
+    }
     // TODO currently SignalWithStartWorkflowExecutionResponse doesn't have eagerWorkflowTask.
     //  We should wire it when it's implemented server-side.
     return new WorkflowSignalWithStartOutput(new WorkflowStartOutput(execution));

temporal-sdk/src/main/java/io/temporal/internal/client/external/GenericWorkflowClient.java

@@ -10,7 +10,7 @@ public interface GenericWorkflowClient {
 
   StartWorkflowExecutionResponse start(StartWorkflowExecutionRequest request);
 
-  void signal(SignalWorkflowExecutionRequest request);
+  SignalWorkflowExecutionResponse signal(SignalWorkflowExecutionRequest request);
 
   SignalWithStartWorkflowExecutionResponse signalWithStart(
       SignalWithStartWorkflowExecutionRequest request);

temporal-sdk/src/main/java/io/temporal/internal/client/external/GenericWorkflowClientImpl.java

@@ -61,13 +61,13 @@ private static Map<String, String> tagsForStartWorkflow(StartWorkflowExecutionRe
   }
 
   @Override
-  public void signal(SignalWorkflowExecutionRequest request) {
+  public SignalWorkflowExecutionResponse signal(SignalWorkflowExecutionRequest request) {
     Map<String, String> tags =
         new ImmutableMap.Builder<String, String>(1)
             .put(MetricsTag.SIGNAL_NAME, request.getSignalName())
             .build();
     Scope scope = metricsScope.tagged(tags);
-    grpcRetryer.retry(
+    return grpcRetryer.retryWithResult(
         () ->
             service
                 .blockingStub()

temporal-sdk/src/main/java/io/temporal/internal/nexus/InternalNexusOperationContext.java

@@ -6,6 +6,8 @@
 import io.temporal.common.interceptors.NexusOperationOutboundCallsInterceptor;
 import io.temporal.nexus.NexusOperationContext;
 import io.temporal.nexus.NexusOperationInfo;
+import java.util.Collections;
+import java.util.List;
 
 public class InternalNexusOperationContext {
   private final String namespace;
@@ -15,6 +17,16 @@ public class InternalNexusOperationContext {
   private final WorkflowClient client;
   NexusOperationOutboundCallsInterceptor outboundCalls;
   Link startWorkflowResponseLink;
+  // Links extracted from the inbound Nexus task. Stored once at the task-handler boundary so the
+  // workflow client (signal, signalWithStart) can attach them to outgoing requests via
+  // SignalWorkflowExecutionRequest.links, matching the Go SDK's NexusOperationLinksKey ctx value.
+  private List<Link> nexusOperationLinks = Collections.emptyList();
+  // Backlink returned by SignalWorkflowExecutionResponse.link /
+  // SignalWithStartWorkflowExecutionResponse.signal_link.
+  // Populated by the workflow client and consumed by the task handler when building
+  // StartOperationResponse, so the caller workflow gets a link pointing at the signal event on
+  // the callee.
+  private Link signalWorkflowResponseLink;
 
   public InternalNexusOperationContext(
       String namespace,
@@ -68,6 +80,27 @@ public Link getStartWorkflowResponseLink() {
     return startWorkflowResponseLink;
   }
 
+  /**
+   * Set the {@code common.v1.Link}s extracted from the inbound Nexus task so they can be attached
+   * to any signal RPCs issued by the operation handler.
+   */
+  public void setNexusOperationLinks(List<Link> links) {
+    this.nexusOperationLinks = links == null ? Collections.emptyList() : links;
+  }
+
+  /** Links from the inbound Nexus task; empty if none. Never null. */
+  public List<Link> getNexusOperationLinks() {
+    return nexusOperationLinks;
+  }
+
+  public void setSignalWorkflowResponseLink(Link link) {
+    this.signalWorkflowResponseLink = link;
+  }
+
+  public Link getSignalWorkflowResponseLink() {
+    return signalWorkflowResponseLink;
+  }
+
   private class NexusOperationContextImpl implements NexusOperationContext {
     @Override
     public NexusOperationInfo getInfo() {

temporal-sdk/src/main/java/io/temporal/internal/nexus/NexusTaskHandlerImpl.java

@@ -20,6 +20,7 @@
 import io.temporal.failure.CanceledFailure;
 import io.temporal.failure.TemporalFailure;
 import io.temporal.internal.common.InternalUtils;
+import io.temporal.internal.common.LinkConverter;
 import io.temporal.internal.common.NexusUtil;
 import io.temporal.internal.worker.NexusTask;
 import io.temporal.internal.worker.NexusTaskHandler;
@@ -284,6 +285,10 @@ private StartOperationResponse handleStartOperation(
             .setCallbackUrl(task.getCallback())
             .setRequestId(task.getRequestId());
     task.getCallbackHeaderMap().forEach(operationStartDetails::putCallbackHeader);
+    // Stash the inbound links in common.v1.Link form on the operation context so that signal
+    // RPCs issued by the handler (e.g. SignalWithStartWorkflow on the callee) can attach them
+    // to SignalWorkflowExecutionRequest.links.
+    List<io.temporal.api.common.v1.Link> inboundCommonLinks = new ArrayList<>();
     task.getLinksList()
         .forEach(
             link -> {
@@ -296,7 +301,13 @@ private StartOperationResponse handleStartOperation(
                     "Invalid link URL: " + link.getUrl(),
                     e);
               }
+              io.temporal.api.common.v1.Link commonLink =
+                  LinkConverter.nexusLinkToWorkflowEvent(link);
+              if (commonLink != null) {
+                inboundCommonLinks.add(commonLink);
+              }
             });
+    CurrentNexusOperationContext.get().setNexusOperationLinks(inboundCommonLinks);
 
     HandlerInputContent.Builder input =
         HandlerInputContent.newBuilder().setDataStream(task.getPayload().toByteString().newInput());
@@ -307,13 +318,27 @@ private StartOperationResponse handleStartOperation(
       try {
         OperationStartResult<HandlerResultContent> result =
             startOperation(context, operationStartDetails.build(), input.build());
+        // If a signal RPC issued by the handler returned a backlink, propagate it to the caller
+        // so the caller workflow's history event links to the signal event on the callee. Same
+        // backlink applies to both sync and async response variants.
+        io.temporal.api.nexus.v1.Link signalBacklink = null;
+        io.temporal.api.common.v1.Link signalResponseLink =
+            CurrentNexusOperationContext.get().getSignalWorkflowResponseLink();
+        if (signalResponseLink != null && signalResponseLink.hasWorkflowEvent()) {
+          signalBacklink =
+              LinkConverter.workflowEventToNexusLink(signalResponseLink.getWorkflowEvent());
+        }
+
         if (result.isSync()) {
-          startResponseBuilder.setSyncSuccess(
+          StartOperationResponse.Sync.Builder syncBuilder =
               StartOperationResponse.Sync.newBuilder()
-                  .setPayload(Payload.parseFrom(result.getSyncResult().getDataBytes()))
-                  .build());
+                  .setPayload(Payload.parseFrom(result.getSyncResult().getDataBytes()));
+          if (signalBacklink != null) {
+            syncBuilder.addLinks(signalBacklink);
+          }
+          startResponseBuilder.setSyncSuccess(syncBuilder.build());
         } else {
-          startResponseBuilder.setAsyncSuccess(
+          StartOperationResponse.Async.Builder asyncBuilder =
               StartOperationResponse.Async.newBuilder()
                   .setOperationId(result.getAsyncOperationToken())
                   .setOperationToken(result.getAsyncOperationToken())
@@ -325,8 +350,11 @@ private StartOperationResponse handleStartOperation(
                                       .setType(link.getType())
                                       .setUrl(link.getUri().toString())
                                       .build())
-                          .collect(Collectors.toList()))
-                  .build());
+                          .collect(Collectors.toList()));
+          if (signalBacklink != null) {
+            asyncBuilder.addLinks(signalBacklink);
+          }
+          startResponseBuilder.setAsyncSuccess(asyncBuilder.build());
         }
       } catch (OperationException e) {
         throw e;