Support starting multiple activities from a synchronous Nexus operation handler

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

@@ -64,14 +64,16 @@ public StartActivityOutput startActivity(StartActivityInput input) {
     NexusOperationMetadata nexusOperationMetadata =
         nexusContext == null ? null : nexusContext.getNexusOperationMetadata();
 
+    String requestId =
+        nexusOperationMetadata != null
+            ? nexusOperationMetadata.requestId
+            : UUID.randomUUID().toString();
+
     StartActivityExecutionRequest.Builder request =
         StartActivityExecutionRequest.newBuilder()
             .setNamespace(clientOptions.getNamespace())
             .setIdentity(clientOptions.getIdentity())
-            .setRequestId(
-                nexusOperationMetadata == null
-                    ? UUID.randomUUID().toString()
-                    : nexusOperationMetadata.requestId)
+            .setRequestId(requestId)
             .setActivityId(options.getId())
             .setActivityType(ActivityType.newBuilder().setName(input.getActivityType()).build())
             .setTaskQueue(TaskQueue.newBuilder().setName(options.getTaskQueue()).build())
@@ -121,14 +123,28 @@ public StartActivityOutput startActivity(StartActivityInput input) {
     io.temporal.api.common.v1.Header grpcHeader = HeaderUtils.toHeaderGrpc(input.getHeader(), null);
     request.setHeader(grpcHeader);
 
-    if (nexusOperationMetadata != null) {
-      List<Link> protoLinks = nexusContext.getRequestLinks();
+    List<Link> protoLinks = Collections.emptyList();
+    if (nexusContext != null) {
+      // Every activity start made on this operation-handler thread gets the inbound links,
+      // including starts through a raw ActivityClient -- this is deliberately broader than
+      // nexusOperationMetadata below, since links carry no completion semantics, unlike a
+      // completion callback. attach_request_id is set unconditionally too, mirroring on-conflict
+      // handling for every activity start; it's only meaningful for the one guarded call, whose
+      // request ID is the stable, reused one (see the requestId derivation above) -- a bypass-path
+      // start's fresh, uncorrelated ID makes the flag a no-op for it, which is harmless.
+      protoLinks = nexusContext.getRequestLinks();
       request.addAllLinks(protoLinks);
-      request.setOnConflictOptions(
+      io.temporal.api.common.v1.OnConflictOptions.Builder onConflictOptions =
           io.temporal.api.common.v1.OnConflictOptions.newBuilder()
               .setAttachRequestId(true)
-              .setAttachLinks(true)
-              .setAttachCompletionCallbacks(true));
+              .setAttachLinks(true);
+      if (nexusOperationMetadata != null) {
+        onConflictOptions.setAttachCompletionCallbacks(true);
+      }
+      request.setOnConflictOptions(onConflictOptions);
+    }
+
+    if (nexusOperationMetadata != null) {
       // Generate the operation token from the user-supplied activity ID and namespace so the
       // dual OPERATION_ID + OPERATION_TOKEN headers can be injected before the start RPC fires.
       try {
@@ -168,7 +184,7 @@ public StartActivityOutput startActivity(StartActivityInput input) {
       throw e;
     }
 
-    if (nexusOperationMetadata != null && response.hasLink()) {
+    if (nexusContext != null && response.hasLink()) {
       nexusContext.addResponseLink(response.getLink());
     }
 

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

@@ -27,6 +27,12 @@ public class InternalNexusOperationContext {
   // workflow client can attach them to the outgoing requests it issues (e.g. signal,
   // signalWithStart) via the request's links field.
   private List<Link> requestLinks = Collections.emptyList();
+  // The inbound Nexus task's request ID, captured at the task-handler boundary and available to
+  // clients executing on the operation-handler thread. RootActivityClientInvoker reuses it for
+  // redelivery-safe activity-start deduplication. It is deliberately independent of
+  // nexusOperationMetadata, which is scoped to the single backing start because it carries
+  // completion-callback semantics.
+  private String requestId;
   // Links returned by outbound RPCs the operation handler issues (such as
   // SignalWorkflowExecutionResponse.link or SignalWithStartWorkflowExecutionResponse.signal_link).
   // One entry per outbound RPC that returned a link. Drained
@@ -106,6 +112,16 @@ public void setRequestLinks(List<Link> links) {
     return Collections.unmodifiableList(requestLinks);
   }
 
+  /** Set the request ID of the inbound Nexus task, ambient for the whole invocation. */
+  public void setRequestId(String requestId) {
+    this.requestId = requestId;
+  }
+
+  /** The inbound Nexus task's request ID; {@code null} if not set. */
+  public String getRequestId() {
+    return requestId;
+  }
+
   public void setStartWorkflowResponseLink(Link link) {
     this.startWorkflowResponseLink = link;
   }

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

@@ -313,6 +313,9 @@ private StartOperationResponse handleStartOperation(
               }
             });
     CurrentNexusOperationContext.get().setRequestLinks(inboundCommonLinks);
+    // Ambient for the whole operation-handler invocation, independent of NexusOperationMetadata.
+    // see InternalNexusOperationContext.requestId.
+    CurrentNexusOperationContext.get().setRequestId(task.getRequestId());
 
     HandlerInputContent.Builder input =
         HandlerInputContent.newBuilder().setDataStream(task.getPayload().toByteString().newInput());

temporal-sdk/src/test/java/io/temporal/internal/client/RootActivityClientInvokerTest.java

@@ -2,6 +2,7 @@
 
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
@@ -22,6 +23,7 @@
 import java.time.Duration;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import org.junit.After;
 import org.junit.Assert;
@@ -132,8 +134,65 @@ public void nexusMetadataWithEmptyCallbackUrlOmitsCompletionCallback() {
   }
 
   @Test
-  public void nexusContextWithoutMetadataStartsOrdinaryActivity() {
-    nexusContext.setRequestLinks(Collections.singletonList(workflowEventLink()));
+  public void nexusContextWithoutMetadataGetsAmbientLinksButFreshRequestIdAndNoCallback() {
+    Link link = workflowEventLink();
+    nexusContext.setRequestLinks(Collections.singletonList(link));
+    nexusContext.setRequestId("ambient-nexus-request-id");
+
+    invoker.startActivity(newStartActivityInput());
+
+    ArgumentCaptor<StartActivityExecutionRequest> captor =
+        ArgumentCaptor.forClass(StartActivityExecutionRequest.class);
+    verify(genericClient).startActivity(captor.capture());
+    StartActivityExecutionRequest request = captor.getValue();
+    Assert.assertFalse(request.getRequestId().isEmpty());
+    Assert.assertNotEquals("ambient-nexus-request-id", request.getRequestId());
+    Assert.assertEquals(Collections.singletonList(link), request.getLinksList());
+    Assert.assertEquals(0, request.getCompletionCallbacksCount());
+    Assert.assertTrue(request.getOnConflictOptions().getAttachRequestId());
+    Assert.assertTrue(request.getOnConflictOptions().getAttachLinks());
+    Assert.assertFalse(request.getOnConflictOptions().getAttachCompletionCallbacks());
+    Assert.assertEquals(Collections.singletonList(activityLink()), nexusContext.getResponseLinks());
+  }
+
+  @Test
+  public void twoStartsForSameActivityIdGetDistinctFreshRequestIds() {
+    nexusContext.setRequestId("ambient-nexus-request-id");
+
+    invoker.startActivity(newStartActivityInput());
+    invoker.startActivity(newStartActivityInput());
+
+    ArgumentCaptor<StartActivityExecutionRequest> captor =
+        ArgumentCaptor.forClass(StartActivityExecutionRequest.class);
+    verify(genericClient, times(2)).startActivity(captor.capture());
+    List<StartActivityExecutionRequest> requests = captor.getAllValues();
+    String firstRequestId = requests.get(0).getRequestId();
+    String secondRequestId = requests.get(1).getRequestId();
+    Assert.assertNotEquals(firstRequestId, secondRequestId);
+    Assert.assertNotEquals("ambient-nexus-request-id", firstRequestId);
+    Assert.assertNotEquals("ambient-nexus-request-id", secondRequestId);
+  }
+
+  @Test
+  public void nexusContextWithoutAmbientStateStartsOrdinaryActivity() {
+    invoker.startActivity(newStartActivityInput());
+
+    ArgumentCaptor<StartActivityExecutionRequest> captor =
+        ArgumentCaptor.forClass(StartActivityExecutionRequest.class);
+    verify(genericClient).startActivity(captor.capture());
+    StartActivityExecutionRequest request = captor.getValue();
+    Assert.assertFalse(request.getRequestId().isEmpty());
+    Assert.assertEquals(0, request.getLinksCount());
+    Assert.assertEquals(0, request.getCompletionCallbacksCount());
+    Assert.assertTrue(request.getOnConflictOptions().getAttachRequestId());
+    Assert.assertTrue(request.getOnConflictOptions().getAttachLinks());
+    Assert.assertFalse(request.getOnConflictOptions().getAttachCompletionCallbacks());
+    Assert.assertEquals(Collections.singletonList(activityLink()), nexusContext.getResponseLinks());
+  }
+
+  @Test
+  public void outsideNexusContextStartsOrdinaryActivity() {
+    CurrentNexusOperationContext.unset();
 
     invoker.startActivity(newStartActivityInput());