Signal links

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().addSignalWorkflowResponseLink(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().addSignalWorkflowResponseLink(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,9 @@
 import io.temporal.common.interceptors.NexusOperationOutboundCallsInterceptor;
 import io.temporal.nexus.NexusOperationContext;
 import io.temporal.nexus.NexusOperationInfo;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
 
 public class InternalNexusOperationContext {
   private final String namespace;
@@ -15,6 +18,21 @@ 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();
+  // Backlinks returned by SignalWorkflowExecutionResponse.link /
+  // SignalWithStartWorkflowExecutionResponse.signal_link. One entry per signal RPC issued from
+  // within the Nexus operation handler. Drained by the task handler when building
+  // StartOperationResponse so every signal the handler issues gets a corresponding link on the
+  // caller workflow's history event.
+  //
+  // NOTE: this context is only safe for use from the single thread that runs the operation
+  // handler (the Nexus task executor's thread). Handlers that spawn their own threads to issue
+  // signals will not see the thread-local context, so the links from those signals will not
+  // propagate.
+  private final List<Link> signalWorkflowResponseLinks = new ArrayList<>();
 
   public InternalNexusOperationContext(
       String namespace,
@@ -68,6 +86,35 @@ 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;
+  }
+
+  /**
+   * Append a backlink returned by a signal-class RPC (signal or signalWithStart). Each signal the
+   * operation handler issues should add one entry; the task handler drains the list when building
+   * the operation's StartOperationResponse.
+   */
+  public void addSignalWorkflowResponseLink(Link link) {
+    if (link != null) {
+      this.signalWorkflowResponseLinks.add(link);
+    }
+  }
+
+  /** Backlinks from every signal RPC issued by the handler. Never null; may be empty. */
+  public List<Link> getSignalWorkflowResponseLinks() {
+    return signalWorkflowResponseLinks;
+  }
+
   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,23 @@ private StartOperationResponse handleStartOperation(
                     "Invalid link URL: " + link.getUrl(),
                     e);
               }
+              // LinkConverter only returns a WorkflowEvent-shaped common.v1.Link; nexus links of
+              // other shapes (e.g. non-temporal URLs) come back null and are intentionally not
+              // forwarded onto SignalWorkflowExecutionRequest.links, which requires the
+              // WorkflowEvent variant. Log so a debugging session can see what was dropped.
+              io.temporal.api.common.v1.Link commonLink =
+                  LinkConverter.nexusLinkToWorkflowEvent(link);
+              if (commonLink != null) {
+                inboundCommonLinks.add(commonLink);
+              } else {
+                log.warn(
+                    "Dropping inbound Nexus link from outbound signal propagation: type='{}',"
+                        + " url='{}' (not a parseable temporal WorkflowEvent link)",
+                    link.getType(),
+                    link.getUrl());
+              }
             });
+    CurrentNexusOperationContext.get().setNexusOperationLinks(inboundCommonLinks);
 
     HandlerInputContent.Builder input =
         HandlerInputContent.newBuilder().setDataStream(task.getPayload().toByteString().newInput());
@@ -307,10 +328,27 @@ private StartOperationResponse handleStartOperation(
       try {
         OperationStartResult<HandlerResultContent> result =
             startOperation(context, operationStartDetails.build(), input.build());
+        // If signal/signalWithStart RPCs issued by the handler returned backlinks, propagate
+        // them to the caller so the caller workflow's history event links to each signal event
+        // on the callee. Same set of backlinks applies to both sync and async response variants.
+        List<io.temporal.api.nexus.v1.Link> signalBacklinks = new ArrayList<>();
+        for (io.temporal.api.common.v1.Link signalResponseLink :
+            CurrentNexusOperationContext.get().getSignalWorkflowResponseLinks()) {
+          if (!signalResponseLink.hasWorkflowEvent()) {
+            continue;
+          }
+          io.temporal.api.nexus.v1.Link converted =
+              LinkConverter.workflowEventToNexusLink(signalResponseLink.getWorkflowEvent());
+          if (converted != null) {
+            signalBacklinks.add(converted);
+          }
+        }
+
         if (result.isSync()) {
           startResponseBuilder.setSyncSuccess(
               StartOperationResponse.Sync.newBuilder()
                   .setPayload(Payload.parseFrom(result.getSyncResult().getDataBytes()))
+                  .addAllLinks(signalBacklinks)
                   .build());
         } else {
           startResponseBuilder.setAsyncSuccess(
@@ -326,6 +364,7 @@ private StartOperationResponse handleStartOperation(
                                       .setUrl(link.getUri().toString())
                                       .build())
                           .collect(Collectors.toList()))
+                  .addAllLinks(signalBacklinks)
                   .build());
         }
       } catch (OperationException e) {