Propagate Nexus serialization context to workflows and standalone activities

CHANGELOG.md

@@ -51,6 +51,13 @@ to docs, or any other relevant information.
   `TEMPORAL_WORKFLOW_TASK_DURATION_WARN_SECONDS` environment variable.
 - Nexus operation handlers now forward inbound callback links to the workflows and signals they start, instead of
   dropping them.
+- Nexus handlers now propagate `NexusSerializationContext` to the workflows and standalone
+  activities they start. These use it only for their results and failures, which are returned to
+  the Nexus caller. Standalone activities also use it for their heartbeat details, which the server
+  includes in their timeout failures.
+  To complete such an activity asynchronously, read the context from
+  `ActivityInfo.getNexusSerializationContext()`, store it with the task token, and pass it to
+  `ActivityCompletionClient.withContext(NexusSerializationContext)`.
 
 ### Changed
 - Moved the Spring AI integration to [temporalio/ai-integrations](https://github.com/temporalio/ai-integrations/tree/main/java/spring-ai).

temporal-sdk/src/main/java/io/temporal/activity/ActivityInfo.java

@@ -1,8 +1,10 @@
 package io.temporal.activity;
 
 import io.temporal.api.common.v1.Payloads;
+import io.temporal.common.Experimental;
 import io.temporal.common.Priority;
 import io.temporal.common.RetryOptions;
+import io.temporal.payload.context.NexusSerializationContext;
 import java.time.Duration;
 import java.util.Optional;
 import javax.annotation.Nonnull;
@@ -171,4 +173,19 @@ default boolean isInWorkflow() {
    * @return Retry options for the Activity Execution.
    */
   RetryOptions getRetryOptions();
+
+  /**
+   * Returns the context of the Nexus operation handler that started this standalone activity, or
+   * {@code null} if it wasn't started by one.
+   *
+   * <p>To complete the activity asynchronously, store this context together with the {@link
+   * #getTaskToken() task token} and pass it to {@link
+   * io.temporal.client.ActivityCompletionClient#withContext(NexusSerializationContext)}, so the
+   * result is encoded with the context the Nexus caller decodes it with.
+   */
+  @Experimental
+  @Nullable
+  default NexusSerializationContext getNexusSerializationContext() {
+    return null;
+  }
 }

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

@@ -1,11 +1,13 @@
 package io.temporal.client;
 
 import io.temporal.activity.ActivityExecutionContext;
+import io.temporal.activity.ActivityInfo;
 import io.temporal.common.Experimental;
 import io.temporal.common.converter.DataConverter;
 import io.temporal.common.converter.PayloadConverter;
 import io.temporal.payload.codec.PayloadCodec;
 import io.temporal.payload.context.ActivitySerializationContext;
+import io.temporal.payload.context.NexusSerializationContext;
 import io.temporal.payload.context.SerializationContext;
 import java.util.Optional;
 import javax.annotation.Nonnull;
@@ -188,4 +190,19 @@ <V> void heartbeatStandalone(String activityId, Optional<String> activityRunId,
   @Experimental
   @Nonnull
   ActivityCompletionClient withContext(@Nonnull ActivitySerializationContext context);
+
+  /**
+   * Supply this context when completing a standalone activity started by a Nexus operation handler,
+   * so its result, failure and heartbeat details are encoded with the context the Nexus caller
+   * decodes them with. Get it from {@link ActivityInfo#getNexusSerializationContext()} and store it
+   * together with the task token.
+   *
+   * @param context the Nexus context of the operation handler that started the activity
+   * @return an instance of ActivityCompletionClient that uses the provided {@code context} for
+   *     serialization
+   * @see NexusSerializationContext
+   */
+  @Experimental
+  @Nonnull
+  ActivityCompletionClient withContext(@Nonnull NexusSerializationContext context);
 }

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

@@ -4,6 +4,8 @@
 import io.temporal.api.common.v1.WorkflowExecution;
 import io.temporal.internal.client.external.ManualActivityCompletionClientFactory;
 import io.temporal.payload.context.ActivitySerializationContext;
+import io.temporal.payload.context.NexusSerializationContext;
+import io.temporal.payload.context.SerializationContext;
 import io.temporal.workflow.Functions;
 import java.util.Optional;
 import javax.annotation.Nonnull;
@@ -14,13 +16,13 @@ class ActivityCompletionClientImpl implements ActivityCompletionClient {
   private final ManualActivityCompletionClientFactory factory;
   private final Functions.Proc completionHandle;
   private final Scope metricsScope;
-  private final @Nullable ActivitySerializationContext serializationContext;
+  private final @Nullable SerializationContext serializationContext;
 
   ActivityCompletionClientImpl(
       ManualActivityCompletionClientFactory manualActivityCompletionClientFactory,
       Functions.Proc completionHandle,
       Scope metricsScope,
-      @Nullable ActivitySerializationContext serializationContext) {
+      @Nullable SerializationContext serializationContext) {
     this.factory = manualActivityCompletionClientFactory;
     this.completionHandle = completionHandle;
     this.metricsScope = metricsScope;
@@ -146,6 +148,12 @@ public ActivityCompletionClient withContext(@Nonnull ActivitySerializationContex
     return new ActivityCompletionClientImpl(factory, completionHandle, metricsScope, context);
   }
 
+  @Nonnull
+  @Override
+  public ActivityCompletionClient withContext(@Nonnull NexusSerializationContext context) {
+    return new ActivityCompletionClientImpl(factory, completionHandle, metricsScope, context);
+  }
+
   @Override
   public <V> void heartbeatStandalone(String activityId, Optional<String> activityRunId, V details)
       throws ActivityCompletionException {

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

@@ -13,6 +13,7 @@
 import io.temporal.internal.common.ProtobufTimeUtils;
 import io.temporal.internal.common.RetryOptionsUtils;
 import io.temporal.internal.common.SearchAttributesUtil;
+import io.temporal.internal.nexus.PropagatedNexusSerializationContext;
 import io.temporal.payload.context.ActivitySerializationContext;
 import java.lang.reflect.Type;
 import java.time.Duration;
@@ -29,6 +30,7 @@ public final class ActivityExecutionDescription extends ActivityExecutionMetadat
 
   private final DescribeActivityExecutionResponse response;
   private final DataConverter dataConverter;
+  private final DataConverter resultDataConverter;
 
   public ActivityExecutionDescription(
       DescribeActivityExecutionResponse response, DataConverter dataConverter, String namespace) {
@@ -54,6 +56,16 @@ public ActivityExecutionDescription(
         dataConverter.withContext(
             new ActivitySerializationContext(
                 namespace, null, null, getActivityType(), getTaskQueue(), false));
+    // An activity started from a Nexus operation handler returns its result and failure to the
+    // Nexus caller, so they are serialized with the handler's Nexus context, which the server
+    // returns with the activity info. Heartbeat details use it too, because the server puts them
+    // into the timeout failure. Input and user metadata keep the activity context.
+    this.resultDataConverter =
+        response.getInfo().hasPropagatedNexusSerializationContext()
+            ? dataConverter.withContext(
+                PropagatedNexusSerializationContext.fromProto(
+                    response.getInfo().getPropagatedNexusSerializationContext()))
+            : this.dataConverter;
   }
 
   private static @Nullable String nullIfEmpty(String s) {
@@ -177,7 +189,7 @@ public boolean hasLastFailure() {
   @Nullable
   public RuntimeException getLastFailure() {
     return response.getInfo().hasLastFailure()
-        ? dataConverter.failureToException(response.getInfo().getLastFailure())
+        ? resultDataConverter.failureToException(response.getInfo().getLastFailure())
         : null;
   }
 
@@ -252,7 +264,8 @@ public boolean hasHeartbeatDetails() {
    * requested without {@link DescribeActivityOptions.Builder#setIncludeHeartbeatDetails(boolean)}.
    */
   public EncodedValues getHeartbeatDetails() {
-    return new EncodedValues(Optional.of(response.getInfo().getHeartbeatDetails()), dataConverter);
+    return new EncodedValues(
+        Optional.of(response.getInfo().getHeartbeatDetails()), resultDataConverter);
   }
 
   /**
@@ -303,7 +316,7 @@ public <V> Optional<V> getResult(Class<V> valueType, @Nullable Type genericType)
       return Optional.empty();
     }
     return Optional.ofNullable(
-        dataConverter.fromPayloads(
+        resultDataConverter.fromPayloads(
             0,
             Optional.of(response.getOutcome().getResult()),
             valueType,
@@ -322,7 +335,7 @@ public RuntimeException getOutcomeFailure() {
     if (!response.getOutcome().hasFailure()) {
       return null;
     }
-    return dataConverter.failureToException(response.getOutcome().getFailure());
+    return resultDataConverter.failureToException(response.getOutcome().getFailure());
   }
 
   /**

temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityExecutionContextImpl.java

@@ -11,7 +11,6 @@
 import io.temporal.common.converter.DataConverter;
 import io.temporal.internal.client.external.ManualActivityCompletionClientFactory;
 import io.temporal.internal.payload.storage.ExternalStorageRunner;
-import io.temporal.payload.context.ActivitySerializationContext;
 import io.temporal.workflow.Functions;
 import java.lang.reflect.Type;
 import java.time.Duration;
@@ -155,13 +154,11 @@ public ManualActivityCompletionClient useLocalManualCompletion() {
     try {
       doNotCompleteOnReturn();
       useLocalManualCompletion = true;
-      ActivitySerializationContext activitySerializationContext =
-          new ActivitySerializationContext(info);
       return new CompletionAwareManualCompletionClient(
           manualCompletionClientFactory.getClient(
               info.getTaskToken(),
               metricsScope,
-              activitySerializationContext,
+              HeartbeatContextImpl.heartbeatContext(info.getNamespace(), info),
               HeartbeatContextImpl.storageTargetForActivity(info.getNamespace(), info)),
           completionHandle);
     } finally {

temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityInfoImpl.java

@@ -9,12 +9,15 @@
 import io.temporal.internal.common.ProtoConverters;
 import io.temporal.internal.common.ProtobufTimeUtils;
 import io.temporal.internal.common.RetryOptionsUtils;
+import io.temporal.internal.nexus.PropagatedNexusSerializationContext;
+import io.temporal.payload.context.NexusSerializationContext;
 import io.temporal.workflow.Functions;
 import java.time.Duration;
 import java.util.Base64;
 import java.util.Objects;
 import java.util.Optional;
 import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
 
 final class ActivityInfoImpl implements ActivityInfoInternal {
   private final String namespace;
@@ -205,6 +208,15 @@ public Optional<Header> getHeader() {
     return Optional.empty();
   }
 
+  @Nullable
+  @Override
+  public NexusSerializationContext getNexusSerializationContext() {
+    return response.hasPropagatedNexusSerializationContext()
+        ? PropagatedNexusSerializationContext.fromProto(
+            response.getPropagatedNexusSerializationContext())
+        : null;
+  }
+
   @Override
   public String toString() {
     return "ActivityInfo{"

temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityTaskExecutors.java

@@ -19,6 +19,7 @@
 import io.temporal.internal.common.FailureUtils;
 import io.temporal.internal.worker.ActivityTaskHandler;
 import io.temporal.payload.context.ActivitySerializationContext;
+import io.temporal.payload.context.NexusSerializationContext;
 import io.temporal.serviceclient.CheckedExceptionWrapper;
 import java.lang.reflect.Method;
 import java.util.HashMap;
@@ -69,6 +70,14 @@ public ActivityTaskHandler.Result execute(ActivityInfoInternal info, Scope metri
               activityInfo.isLocal());
       DataConverter dataConverterWithActivityContext =
           dataConverter.withContext(serializationContext);
+      // The result and failure of a standalone activity started from a Nexus operation handler are
+      // returned to the Nexus caller, so they use the handler's Nexus context. Heartbeat details
+      // use it too, because the server puts them into the timeout failure.
+      NexusSerializationContext nexusContext = info.getNexusSerializationContext();
+      DataConverter resultDataConverter =
+          nexusContext != null
+              ? dataConverter.withContext(nexusContext)
+              : dataConverterWithActivityContext;
 
       try {
         info.getHeader()
@@ -92,7 +101,7 @@ public ActivityTaskHandler.Result execute(ActivityInfoInternal info, Scope metri
               info.getActivityId(), null, null, null, context.isUseLocalManualCompletion());
         }
 
-        return this.constructSuccessfulResultValue(info, result, dataConverterWithActivityContext);
+        return this.constructSuccessfulResultValue(info, result, resultDataConverter);
       } catch (Throwable e) {
         Throwable ex = CheckedExceptionWrapper.unwrap(e);
         boolean local = info.isLocal();
@@ -127,7 +136,7 @@ public ActivityTaskHandler.Result execute(ActivityInfoInternal info, Scope metri
             context.getLastHeartbeatValue(),
             metricsScope,
             local,
-            dataConverterWithActivityContext);
+            resultDataConverter);
       } finally {
         if (context.isDoNotCompleteOnReturn()) {
           if (!context.isUseLocalManualCompletion()) {
@@ -150,19 +159,16 @@ abstract Object[] provideArgs(
         Optional<Payloads> input, DataConverter dataConverterWithActivityContext);
 
     protected abstract ActivityTaskHandler.Result constructSuccessfulResultValue(
-        ActivityInfoInternal info,
-        ActivityOutput result,
-        DataConverter dataConverterWithActivityContext);
+        ActivityInfoInternal info, ActivityOutput result, DataConverter resultDataConverter);
 
     ActivityTaskHandler.Result constructResultValue(
         ActivityInfoInternal info,
         @Nullable ActivityOutput result,
-        DataConverter dataConverterWithActivityContext) {
+        DataConverter resultDataConverter) {
       RespondActivityTaskCompletedRequest.Builder request =
           RespondActivityTaskCompletedRequest.newBuilder();
       if (result != null) {
-        Optional<Payloads> serialized =
-            dataConverterWithActivityContext.toPayloads(result.getResult());
+        Optional<Payloads> serialized = resultDataConverter.toPayloads(result.getResult());
         serialized.ifPresent(request::setResult);
       }
       return new ActivityTaskHandler.Result(
@@ -218,14 +224,12 @@ Object[] provideArgs(Optional<Payloads> input, DataConverter dataConverterWithAc
 
     @Override
     protected ActivityTaskHandler.Result constructSuccessfulResultValue(
-        ActivityInfoInternal info,
-        ActivityOutput result,
-        DataConverter dataConverterWithActivityContext) {
+        ActivityInfoInternal info, ActivityOutput result, DataConverter resultDataConverter) {
       return constructResultValue(
           info,
           // if the expected result of the method is null, we don't publish result at all
           method.getReturnType() != Void.TYPE ? result : null,
-          dataConverterWithActivityContext);
+          resultDataConverter);
     }
   }
 
@@ -261,10 +265,8 @@ Object[] provideArgs(Optional<Payloads> input, DataConverter dataConverterWithAc
 
     @Override
     protected ActivityTaskHandler.Result constructSuccessfulResultValue(
-        ActivityInfoInternal info,
-        ActivityOutput result,
-        DataConverter dataConverterWithActivityContext) {
-      return constructResultValue(info, result, dataConverterWithActivityContext);
+        ActivityInfoInternal info, ActivityOutput result, DataConverter resultDataConverter) {
+      return constructResultValue(info, result, resultDataConverter);
     }
   }
 }