Add Standalone Activities to Temporal Nexus Operation Handler

temporal-sdk/src/main/java/io/temporal/client/ActivityClientImpl.java

@@ -6,6 +6,7 @@
 import io.temporal.common.interceptors.ActivityClientCallsInterceptor;
 import io.temporal.common.interceptors.ActivityClientInterceptor;
 import io.temporal.common.interceptors.Header;
+import io.temporal.internal.client.ActivityClientInternal;
 import io.temporal.internal.client.ActivityHandleImpl;
 import io.temporal.internal.client.RootActivityClientInvoker;
 import io.temporal.internal.client.external.GenericWorkflowClientImpl;
@@ -27,7 +28,7 @@
  * Implementation of {@link ActivityClient} that delegates calls through the activity interceptor
  * chain and ultimately to the Temporal service.
  */
-class ActivityClientImpl implements ActivityClient {
+class ActivityClientImpl implements ActivityClient, ActivityClientInternal {
 
   private final WorkflowServiceStubs stubs;
   private final ActivityClientOptions options;
@@ -56,6 +57,11 @@ private static ActivityClientCallsInterceptor initializeClientInvoker(
     return invoker;
   }
 
+  @Override
+  public ActivityClientCallsInterceptor getInvoker() {
+    return invoker;
+  }
+
   // ---- Interface-based start (Proc variants) ----
 
   @Override

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

@@ -0,0 +1,16 @@
+package io.temporal.internal.client;
+
+import io.temporal.common.interceptors.ActivityClientCallsInterceptor;
+
+/**
+ * Internal-only view of an {@code ActivityClient} that exposes the configured interceptor chain.
+ *
+ * <p>Lives in {@code io.temporal.internal.client} so that other internal SDK packages (e.g. {@code
+ * io.temporal.nexus}) can route a fully-constructed {@link
+ * ActivityClientCallsInterceptor.StartActivityInput} through the chain without bypassing
+ * user-registered interceptors or the metrics-tagged scope, and without forcing the concrete impl
+ * class to be public.
+ */
+public interface ActivityClientInternal {
+  ActivityClientCallsInterceptor getInvoker();
+}

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

@@ -0,0 +1,82 @@
+package io.temporal.internal.client;
+
+import io.nexusrpc.Link;
+import io.temporal.client.StartActivityOptions;
+import io.temporal.common.Experimental;
+import io.temporal.common.interceptors.Header;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Request used to start an activity from a Nexus operation handler. Mirrors {@link
+ * NexusStartWorkflowRequest} but carries the activity-specific scheduling payload.
+ */
+@Experimental
+public final class NexusStartActivityRequest {
+  private final String requestId;
+  private final String callbackUrl;
+  private final Map<String, String> callbackHeaders;
+  private final String taskQueue;
+  private final List<Link> links;
+  private final String activityType;
+  private final List<Object> args;
+  private final StartActivityOptions options;
+  private final Header header;
+
+  public NexusStartActivityRequest(
+      String requestId,
+      String callbackUrl,
+      Map<String, String> callbackHeaders,
+      String taskQueue,
+      List<Link> links,
+      String activityType,
+      List<Object> args,
+      StartActivityOptions options,
+      Header header) {
+    this.requestId = requestId;
+    this.callbackUrl = callbackUrl;
+    this.callbackHeaders = callbackHeaders;
+    this.taskQueue = taskQueue;
+    this.links = links;
+    this.activityType = activityType;
+    this.args = args;
+    this.options = options;
+    this.header = header;
+  }
+
+  public String getRequestId() {
+    return requestId;
+  }
+
+  public String getCallbackUrl() {
+    return callbackUrl;
+  }
+
+  public Map<String, String> getCallbackHeaders() {
+    return callbackHeaders;
+  }
+
+  public String getTaskQueue() {
+    return taskQueue;
+  }
+
+  public List<Link> getLinks() {
+    return links;
+  }
+
+  public String getActivityType() {
+    return activityType;
+  }
+
+  public List<Object> getArgs() {
+    return args;
+  }
+
+  public StartActivityOptions getOptions() {
+    return options;
+  }
+
+  public Header getHeader() {
+    return header;
+  }
+}

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

@@ -0,0 +1,35 @@
+package io.temporal.internal.client;
+
+import io.temporal.common.Experimental;
+import javax.annotation.Nullable;
+
+/**
+ * Response returned from starting an activity via {@link NexusStartActivityRequest}. Mirrors {@link
+ * NexusStartWorkflowResponse}.
+ */
+@Experimental
+public final class NexusStartActivityResponse {
+  private final String activityId;
+  private final @Nullable String runId;
+  private final String operationToken;
+
+  public NexusStartActivityResponse(
+      String activityId, @Nullable String runId, String operationToken) {
+    this.activityId = activityId;
+    this.runId = runId;
+    this.operationToken = operationToken;
+  }
+
+  public String getActivityId() {
+    return activityId;
+  }
+
+  @Nullable
+  public String getRunId() {
+    return runId;
+  }
+
+  public String getOperationToken() {
+    return operationToken;
+  }
+}

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

@@ -3,12 +3,15 @@
 import static io.temporal.internal.common.RetryOptionsUtils.toRetryPolicy;
 import static io.temporal.internal.common.WorkflowExecutionUtils.makeUserMetaData;
 
+import com.google.common.base.Strings;
 import com.google.common.collect.Iterators;
 import io.grpc.Deadline;
 import io.grpc.Status;
 import io.grpc.StatusRuntimeException;
 import io.temporal.api.activity.v1.ActivityExecutionOutcome;
 import io.temporal.api.common.v1.ActivityType;
+import io.temporal.api.common.v1.Callback;
+import io.temporal.api.common.v1.Link;
 import io.temporal.api.common.v1.Payloads;
 import io.temporal.api.errordetails.v1.ActivityExecutionAlreadyStartedFailure;
 import io.temporal.api.sdk.v1.UserMetadata;
@@ -19,9 +22,13 @@
 import io.temporal.common.interceptors.ActivityClientCallsInterceptor;
 import io.temporal.internal.client.external.GenericWorkflowClient;
 import io.temporal.internal.common.HeaderUtils;
+import io.temporal.internal.common.InternalUtils;
 import io.temporal.internal.common.ProtoConverters;
 import io.temporal.internal.common.ProtobufTimeUtils;
 import io.temporal.internal.common.SearchAttributesUtil;
+import io.temporal.internal.nexus.CurrentNexusOperationContext;
+import io.temporal.internal.nexus.InternalNexusOperationContext;
+import io.temporal.internal.nexus.NexusOperationMetadata;
 import io.temporal.serviceclient.StatusUtils;
 import java.lang.reflect.Type;
 import java.util.*;
@@ -49,12 +56,19 @@ public RootActivityClientInvoker(
   public StartActivityOutput startActivity(StartActivityInput input) {
     StartActivityOptions options = input.getOptions();
     DataConverter dc = clientOptions.getDataConverter();
+    InternalNexusOperationContext nexusContext =
+        CurrentNexusOperationContext.isNexusContext() ? CurrentNexusOperationContext.get() : null;
+    NexusOperationMetadata nexusOperationMetadata =
+        nexusContext == null ? null : nexusContext.getNexusOperationMetadata();
 
     StartActivityExecutionRequest.Builder request =
         StartActivityExecutionRequest.newBuilder()
             .setNamespace(clientOptions.getNamespace())
             .setIdentity(clientOptions.getIdentity())
-            .setRequestId(UUID.randomUUID().toString())
+            .setRequestId(
+                nexusOperationMetadata == null
+                    ? UUID.randomUUID().toString()
+                    : nexusOperationMetadata.requestId)
             .setActivityId(options.getId())
             .setActivityType(ActivityType.newBuilder().setName(input.getActivityType()).build())
             .setTaskQueue(TaskQueue.newBuilder().setName(options.getTaskQueue()).build())
@@ -104,6 +118,32 @@ 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();
+      request.addAllLinks(protoLinks);
+      // 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 {
+        nexusOperationMetadata.operationToken =
+            io.temporal.internal.nexus.OperationTokenUtil.generateActivityExecutionOperationToken(
+                options.getId(), clientOptions.getNamespace());
+      } catch (com.fasterxml.jackson.core.JsonProcessingException e) {
+        throw new io.nexusrpc.handler.HandlerException(
+            io.nexusrpc.handler.HandlerException.ErrorType.BAD_REQUEST,
+            "failed to generate activity operation token",
+            e);
+      }
+      if (!Strings.isNullOrEmpty(nexusOperationMetadata.callbackUrl)) {
+        Callback cb =
+            InternalUtils.buildNexusCallback(
+                nexusOperationMetadata.callbackUrl,
+                nexusOperationMetadata.callbackHeaders,
+                nexusOperationMetadata.operationToken,
+                protoLinks);
+        request.addCompletionCallbacks(cb);
+      }
+    }
+
     StartActivityExecutionResponse response;
     try {
       response = genericClient.startActivity(request.build());
@@ -120,6 +160,10 @@ public StartActivityOutput startActivity(StartActivityInput input) {
       throw e;
     }
 
+    if (nexusOperationMetadata != null && response.hasLink()) {
+      nexusContext.addResponseLink(response.getLink());
+    }
+
     String runId = response.getRunId().isEmpty() ? null : response.getRunId();
     return new StartActivityOutput(options.getId(), runId);
   }

temporal-sdk/src/main/java/io/temporal/internal/common/InternalUtils.java

@@ -104,35 +104,10 @@ public static NexusWorkflowStarter createNexusBoundStub(
 
     // If a callback URL is provided, pass it as a completion callback.
     if (!Strings.isNullOrEmpty(request.getCallbackUrl())) {
-      // Add the Nexus operation ID to the headers if it is not already present to support
-      // fabricating
-      // a NexusOperationStarted event if the completion is received before the response to a
-      // StartOperation request.
-      Map<String, String> headers =
-          request.getCallbackHeaders().entrySet().stream()
-              .collect(
-                  Collectors.toMap(
-                      (k) -> k.getKey().toLowerCase(),
-                      Map.Entry::getValue,
-                      (a, b) -> a,
-                      () -> new TreeMap<>(String.CASE_INSENSITIVE_ORDER)));
-      if (!headers.containsKey(Header.OPERATION_ID)) {
-        headers.put(Header.OPERATION_ID.toLowerCase(), operationToken);
-      }
-      if (!headers.containsKey(Header.OPERATION_TOKEN)) {
-        headers.put(Header.OPERATION_TOKEN.toLowerCase(), operationToken);
-      }
-      Callback.Builder cbBuilder =
-          Callback.newBuilder()
-              .setNexus(
-                  Callback.Nexus.newBuilder()
-                      .setUrl(request.getCallbackUrl())
-                      .putAllHeader(headers)
-                      .build());
-      if (links != null) {
-        cbBuilder.addAllLinks(links);
-      }
-      nexusWorkflowOptions.setCompletionCallbacks(Collections.singletonList(cbBuilder.build()));
+      Callback cb =
+          buildNexusCallback(
+              request.getCallbackUrl(), request.getCallbackHeaders(), operationToken, links);
+      nexusWorkflowOptions.setCompletionCallbacks(Collections.singletonList(cb));
     }
 
     if (options.getTaskQueue() == null) {
@@ -157,6 +132,47 @@ public static boolean isWorkflowStreamReservedName(String name) {
     return name.startsWith(WORKFLOW_STREAM_RESERVED_PREFIX);
   }
 
+  /**
+   * Builds a {@link Callback} for use as a Nexus completion callback. Injects both the legacy
+   * {@code Nexus-Operation-Id} and the newer {@code Nexus-Operation-Token} headers
+   * (case-insensitive lookup) when not already present so the server can fabricate
+   * operation-started events if the completion is received before the response to a StartOperation
+   * request.
+   *
+   * <p>Shared by the workflow start path ({@link #createNexusBoundStub}) and the activity start
+   * path ({@code RootActivityClientInvoker.startActivity}). The dual {@code OPERATION_ID} + {@code
+   * OPERATION_TOKEN} headers must be injected before the start RPC is issued.
+   */
+  @SuppressWarnings("deprecation") // Check the OPERATION_ID header for backwards compatibility
+  public static Callback buildNexusCallback(
+      String callbackUrl,
+      Map<String, String> callbackHeaders,
+      String operationToken,
+      List<Link> protoLinks) {
+    Map<String, String> headers =
+        callbackHeaders.entrySet().stream()
+            .collect(
+                Collectors.toMap(
+                    (k) -> k.getKey().toLowerCase(),
+                    Map.Entry::getValue,
+                    (a, b) -> a,
+                    () -> new TreeMap<>(String.CASE_INSENSITIVE_ORDER)));
+    if (!headers.containsKey(Header.OPERATION_ID)) {
+      headers.put(Header.OPERATION_ID.toLowerCase(), operationToken);
+    }
+    if (!headers.containsKey(Header.OPERATION_TOKEN)) {
+      headers.put(Header.OPERATION_TOKEN.toLowerCase(), operationToken);
+    }
+    Callback.Builder cbBuilder =
+        Callback.newBuilder()
+            .setNexus(
+                Callback.Nexus.newBuilder().setUrl(callbackUrl).putAllHeader(headers).build());
+    if (protoLinks != null) {
+      cbBuilder.addAllLinks(protoLinks);
+    }
+    return cbBuilder.build();
+  }
+
   /** Check the method name for reserved prefixes or names. */
   public static void checkMethodName(POJOWorkflowMethodMetadata methodMetadata) {
     boolean workflowStreamExempt =