Add Copyable for local activity transfers

CHANGELOG.md

@@ -20,6 +20,9 @@ to docs, or any other relevant information.
 ## [Unreleased]
 
 ### Added
+- Added experimental `Copyable<T>` for isolated in-process copies. Local activities copy
+  eligible inputs for each attempt and outputs for live execution, preserving history serialization
+  and replay. Deeply immutable values may return themselves.
 - Added experimental `ChildWorkflowOptions.Builder.setVersioningOverride` and
   `VersioningOverride.OneTimeVersioningOverride` for explicit pinned, auto-upgrade, and one-time
   child workflow routing. Invalid child overrides are reported as `InvalidVersioningOverrideFailure`

temporal-sdk/src/main/java/io/temporal/common/Copyable.java

@@ -0,0 +1,30 @@
+package io.temporal.common;
+
+/**
+ * Supports isolated in-process value transfer without a serialization round trip.
+ *
+ * <p>Copies must not share mutable reachable state with the original. Deeply immutable values may
+ * return {@code this}. A copy must be non-null and compatible with the original's runtime type.
+ *
+ * <p>Copying must be side-effect-free, safe for concurrent calls, and deterministically equivalent
+ * to a round trip through the configured data converter. The SDK may call it on workflow or
+ * activity threads, including during replay. Object identity must not affect workflow behavior.
+ *
+ * <p>Currently, only local activities use this interface. Input batches use scheduling-time
+ * snapshots when every non-null argument is copyable. Compatible requested types use fresh copies;
+ * other types use the data converter. Copying is independent of marker options; omitted inputs
+ * avoid serialization unless conversion or explicit payload access requires it. Outputs remain
+ * serialized for history and deserialized on replay. Other execution paths, including eager
+ * workflows and eager activities, do not yet use this interface.
+ *
+ * @param <T> the type of the copied value.
+ */
+@Experimental
+public interface Copyable<T extends Copyable<T>> {
+  /**
+   * Returns an independent copy satisfying this interface's contract.
+   *
+   * @return a non-null copy, or {@code this} for a deeply immutable value.
+   */
+  T copy();
+}

temporal-sdk/src/main/java/io/temporal/common/converter/EncodedValues.java

@@ -5,7 +5,7 @@
 import java.util.Objects;
 import java.util.Optional;
 
-public final class EncodedValues implements Values {
+public class EncodedValues implements Values {
   private Optional<Payloads> payloads;
   private DataConverter converter;
   private final Object[] values;

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

@@ -15,25 +15,29 @@
 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;
   private final String activityTaskQueue;
   private final PollActivityTaskQueueResponseOrBuilder response;
   private final boolean local;
   private final Functions.Proc completionHandle;
+  private final ActivityInput input;
 
   ActivityInfoImpl(
       PollActivityTaskQueueResponseOrBuilder response,
       @Nonnull String namespace,
       @Nonnull String activityTaskQueue,
       boolean local,
-      Functions.Proc completionHandle) {
+      Functions.Proc completionHandle,
+      @Nullable LocalActivityArguments snapshot) {
     this.response = Objects.requireNonNull(response);
     this.namespace = Objects.requireNonNull(namespace);
     this.activityTaskQueue = Objects.requireNonNull(activityTaskQueue);
     this.local = local;
     this.completionHandle = completionHandle;
+    this.input = new ActivityInput(response, snapshot);
   }
 
   @Override
@@ -190,11 +194,8 @@ public Functions.Proc getCompletionHandle() {
   }
 
   @Override
-  public Optional<Payloads> getInput() {
-    if (response.hasInput()) {
-      return Optional.of(response.getInput());
-    }
-    return Optional.empty();
+  public ActivityInput getInput() {
+    return input;
   }
 
   @Override

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

@@ -2,7 +2,6 @@
 
 import io.temporal.activity.ActivityInfo;
 import io.temporal.api.common.v1.Header;
-import io.temporal.api.common.v1.Payloads;
 import io.temporal.workflow.Functions;
 import java.util.Optional;
 
@@ -21,7 +20,7 @@ interface ActivityInfoInternal extends ActivityInfo {
   /**
    * @return input parameters of the activity execution
    */
-  Optional<Payloads> getInput();
+  ActivityInput getInput();
 
   /**
    * @return header that is passed with the activity execution

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

@@ -0,0 +1,81 @@
+package io.temporal.internal.activity;
+
+import io.temporal.api.common.v1.Payloads;
+import io.temporal.api.workflowservice.v1.PollActivityTaskQueueResponseOrBuilder;
+import io.temporal.common.converter.DataConverter;
+import io.temporal.common.converter.EncodedValues;
+import java.lang.reflect.Type;
+import java.util.Objects;
+import java.util.Optional;
+import javax.annotation.Nullable;
+
+/** One source of activity arguments: serialized payloads or a captured copyable snapshot. */
+final class ActivityInput {
+  // A cached serialized view of the input, materialized only if needed.
+  private @Nullable Optional<Payloads> payloads;
+  private final @Nullable LocalActivityArguments snapshot;
+
+  ActivityInput(
+      PollActivityTaskQueueResponseOrBuilder response, @Nullable LocalActivityArguments snapshot) {
+    this.snapshot = snapshot;
+    this.payloads =
+        response.hasInput()
+            ? Optional.of(response.getInput())
+            : snapshot == null ? Optional.empty() : null;
+  }
+
+  Object[] getArguments(Class<?>[] types, Type[] genericTypes, DataConverter converter) {
+    if (snapshot != null && snapshot.canCopyTo(types, genericTypes)) {
+      return snapshot.copy();
+    }
+    return converter.fromPayloads(toPayloads(converter), types, genericTypes);
+  }
+
+  private Optional<Payloads> toPayloads(DataConverter converter) {
+    if (payloads == null) {
+      payloads = snapshot.toPayloads(converter);
+    }
+    return payloads;
+  }
+
+  EncodedValues getEncodedValues(DataConverter converter) {
+    if (snapshot == null) {
+      return new EncodedValues(payloads, converter);
+    }
+    return new CopyableEncodedValues(converter);
+  }
+
+  /**
+   * Reads independent copies without exposing the snapshot used by conversion and serialization.
+   */
+  private final class CopyableEncodedValues extends EncodedValues {
+    private DataConverter converter;
+
+    private CopyableEncodedValues(DataConverter converter) {
+      setDataConverter(converter);
+    }
+
+    @Override
+    public int getSize() {
+      return snapshot.size();
+    }
+
+    @Override
+    public Optional<Payloads> toPayloads() {
+      return ActivityInput.this.toPayloads(converter);
+    }
+
+    @Override
+    public void setDataConverter(DataConverter converter) {
+      this.converter = Objects.requireNonNull(converter);
+    }
+
+    @Override
+    public <T> T get(int index, Class<T> type, Type genericType) {
+      if (snapshot.canCopyTo(index, type, genericType)) {
+        return type.cast(snapshot.copy(index));
+      }
+      return converter.fromPayloads(index, toPayloads(), type, genericType);
+    }
+  }
+}

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

@@ -9,6 +9,6 @@ public static ActivityInfo toActivityInfoImpl(
       String namespace,
       String activityTaskQueue,
       boolean local) {
-    return new ActivityInfoImpl(response, namespace, activityTaskQueue, local, null);
+    return new ActivityInfoImpl(response, namespace, activityTaskQueue, local, null, null);
   }
 }

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

@@ -9,13 +9,14 @@
 import io.temporal.api.common.v1.Payloads;
 import io.temporal.api.workflowservice.v1.RespondActivityTaskCompletedRequest;
 import io.temporal.client.ActivityCanceledException;
+import io.temporal.common.Copyable;
 import io.temporal.common.context.ContextPropagator;
 import io.temporal.common.converter.DataConverter;
-import io.temporal.common.converter.EncodedValues;
 import io.temporal.common.interceptors.ActivityInboundCallsInterceptor;
 import io.temporal.common.interceptors.ActivityInboundCallsInterceptor.ActivityOutput;
 import io.temporal.common.interceptors.Header;
 import io.temporal.common.interceptors.WorkerInterceptor;
+import io.temporal.internal.common.CopyableUtils;
 import io.temporal.internal.common.FailureUtils;
 import io.temporal.internal.worker.ActivityTaskHandler;
 import io.temporal.payload.context.ActivitySerializationContext;
@@ -147,7 +148,7 @@ public ActivityTaskHandler.Result execute(ActivityInfoInternal info, Scope metri
     abstract Object getActivity();
 
     abstract Object[] provideArgs(
-        Optional<Payloads> input, DataConverter dataConverterWithActivityContext);
+        ActivityInput input, DataConverter dataConverterWithActivityContext);
 
     protected abstract ActivityTaskHandler.Result constructSuccessfulResultValue(
         ActivityInfoInternal info,
@@ -160,13 +161,19 @@ ActivityTaskHandler.Result constructResultValue(
         DataConverter dataConverterWithActivityContext) {
       RespondActivityTaskCompletedRequest.Builder request =
           RespondActivityTaskCompletedRequest.newBuilder();
+      Object copiedResult = null;
       if (result != null) {
-        Optional<Payloads> serialized =
-            dataConverterWithActivityContext.toPayloads(result.getResult());
+        Object value = result.getResult();
+        if (info.isLocal() && result.getResult() instanceof Copyable) {
+          copiedResult = CopyableUtils.copy(value);
+          value = copiedResult;
+        }
+        // Serialize the same snapshot delivered to the workflow, not the activity's retained value.
+        Optional<Payloads> serialized = dataConverterWithActivityContext.toPayloads(value);
         serialized.ifPresent(request::setResult);
       }
       return new ActivityTaskHandler.Result(
-          info.getActivityId(), request.build(), null, null, false);
+          info.getActivityId(), request.build(), null, null, false, copiedResult);
     }
 
     static void deserializeAndPopulateContext(
@@ -211,9 +218,11 @@ Object getActivity() {
     }
 
     @Override
-    Object[] provideArgs(Optional<Payloads> input, DataConverter dataConverterWithActivityContext) {
-      return dataConverterWithActivityContext.fromPayloads(
-          input, method.getParameterTypes(), method.getGenericParameterTypes());
+    Object[] provideArgs(ActivityInput input, DataConverter dataConverterWithActivityContext) {
+      return input.getArguments(
+          method.getParameterTypes(),
+          method.getGenericParameterTypes(),
+          dataConverterWithActivityContext);
     }
 
     @Override
@@ -254,9 +263,8 @@ Object getActivity() {
     }
 
     @Override
-    Object[] provideArgs(Optional<Payloads> input, DataConverter dataConverterWithActivityContext) {
-      EncodedValues encodedValues = new EncodedValues(input, dataConverterWithActivityContext);
-      return new Object[] {encodedValues};
+    Object[] provideArgs(ActivityInput input, DataConverter dataConverterWithActivityContext) {
+      return new Object[] {input.getEncodedValues(dataConverterWithActivityContext)};
     }
 
     @Override

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

@@ -103,7 +103,8 @@ public Result handle(ActivityTask activityTask, Scope metricsScope, boolean loca
             this.namespace,
             this.taskQueue,
             localActivity,
-            activityTask.getCompletionCallback());
+            activityTask.getCompletionCallback(),
+            localActivity ? activityTask.getLocalActivityArguments() : null);
     ActivityTaskExecutor activity = activities.get(activityType);
     if (activity != null) {
       return activity.execute(activityInfo, metricsScope);

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

@@ -0,0 +1,82 @@
+package io.temporal.internal.activity;
+
+import io.temporal.api.common.v1.Payloads;
+import io.temporal.common.Copyable;
+import io.temporal.common.converter.DataConverter;
+import io.temporal.internal.common.CopyableUtils;
+import java.lang.reflect.Type;
+import java.util.Optional;
+import javax.annotation.Nullable;
+
+/** A scheduling-time snapshot that is never exposed to an activity attempt. */
+public final class LocalActivityArguments {
+  private final Object[] snapshot;
+
+  private LocalActivityArguments(Object[] snapshot) {
+    this.snapshot = snapshot;
+  }
+
+  /**
+   * Returns a snapshot only when the entire argument list supports copying. Otherwise, the caller
+   * must use the data converter for the complete list to preserve its batch conversion semantics.
+   */
+  @Nullable
+  public static LocalActivityArguments capture(@Nullable Object[] arguments) {
+    if (arguments == null) {
+      arguments = new Object[0];
+    }
+    for (Object argument : arguments) {
+      if (argument != null && !(argument instanceof Copyable)) {
+        return null;
+      }
+    }
+    return new LocalActivityArguments(copy(arguments));
+  }
+
+  /** Returns independent arguments for each attempt, including retries after a workflow timer. */
+  public Object[] copy() {
+    return copy(snapshot);
+  }
+
+  public int size() {
+    return snapshot.length;
+  }
+
+  public Object copy(int index) {
+    return snapshot[index] == null ? null : CopyableUtils.copy(snapshot[index]);
+  }
+
+  public boolean canCopyTo(Class<?>[] parameterTypes, Type[] genericTypes) {
+    if (parameterTypes.length != snapshot.length || genericTypes.length != snapshot.length) {
+      return false;
+    }
+    for (int i = 0; i < snapshot.length; i++) {
+      if (!canCopyTo(i, parameterTypes[i], genericTypes[i])) {
+        return false;
+      }
+    }
+    return true;
+  }
+
+  /** Erased generic types and different target classes require the converter's interpretation. */
+  public boolean canCopyTo(int index, Class<?> type, Type genericType) {
+    return index >= 0
+        && index < snapshot.length
+        && type.equals(genericType)
+        && type.getTypeParameters().length == 0
+        && (snapshot[index] == null ? !type.isPrimitive() : snapshot[index].getClass() == type);
+  }
+
+  /** Preserves conversion behavior for untyped calls with different argument types or counts. */
+  public Optional<Payloads> toPayloads(DataConverter converter) {
+    return converter.toPayloads(snapshot);
+  }
+
+  private static Object[] copy(Object[] arguments) {
+    Object[] copy = new Object[arguments.length];
+    for (int i = 0; i < arguments.length; i++) {
+      copy[i] = arguments[i] == null ? null : CopyableUtils.copy(arguments[i]);
+    }
+    return copy;
+  }
+}

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

@@ -0,0 +1,20 @@
+package io.temporal.internal.common;
+
+import io.temporal.common.Copyable;
+import java.util.Objects;
+
+/** Validates copies before they cross the local activity boundary. */
+public final class CopyableUtils {
+  private CopyableUtils() {}
+
+  public static Object copy(Object value) {
+    Copyable<?> copy =
+        Objects.requireNonNull(
+            ((Copyable<?>) value).copy(), "Copyable.copy() must not return null");
+    if (!value.getClass().isInstance(copy)) {
+      throw new IllegalArgumentException(
+          "Copyable.copy() must return a Copyable compatible with " + value.getClass().getName());
+    }
+    return copy;
+  }
+}