diff --git a/temporal-sdk/src/main/java/io/temporal/internal/payload/storage/ExternalStorageMessageTransformer.java b/temporal-sdk/src/main/java/io/temporal/internal/payload/storage/ExternalStorageMessageTransformer.java
new file mode 100644
index 000000000..7385f9900
--- /dev/null
+++ b/temporal-sdk/src/main/java/io/temporal/internal/payload/storage/ExternalStorageMessageTransformer.java
@@ -0,0 +1,75 @@
+package io.temporal.internal.payload.storage;
+
+import com.google.protobuf.Message;
+import io.temporal.common.CancellationToken;
+import io.temporal.internal.payload.visitor.PayloadVisitorOptions;
+import io.temporal.internal.payload.visitor.PayloadVisitors;
+import io.temporal.payload.storage.StorageDriverTargetInfo;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.CompletableFuture;
+import javax.annotation.Nullable;
+
+/**
+ * Transforms payload lists reachable from a proto message by delegating each visited list to {@link
+ * ExternalStoragePayloadTransformer}.
+ *
+ * <p>Search attributes stay inline because the server indexes and validates their payload values.
+ *
+ * <p>The {@link Message.Builder} overloads transform in place; the {@link Message} overloads copy
+ * through a builder and complete with the copy.
+ */
+final class ExternalStorageMessageTransformer {
+  private final ExternalStoragePayloadTransformer payloadTransformer;
+  private final int payloadVisitConcurrency;
+
+  ExternalStorageMessageTransformer(
+      ExternalStoragePayloadTransformer payloadTransformer, int payloadVisitConcurrency) {
+    this.payloadTransformer = payloadTransformer;
+    this.payloadVisitConcurrency = payloadVisitConcurrency;
+  }
+
+  <T extends Message> CompletableFuture<T> store(
+      T message,
+      @Nullable StorageDriverTargetInfo target,
+      CancellationToken<CancellationException> cancellationToken) {
+    return PayloadVisitors.visit(message, storeOptions(target, cancellationToken));
+  }
+
+  CompletableFuture<Void> store(
+      Message.Builder builder,
+      @Nullable StorageDriverTargetInfo target,
+      CancellationToken<CancellationException> cancellationToken) {
+    return PayloadVisitors.visit(builder, storeOptions(target, cancellationToken));
+  }
+
+  <T extends Message> CompletableFuture<T> retrieve(
+      T message, CancellationToken<CancellationException> cancellationToken) {
+    return PayloadVisitors.visit(message, retrieveOptions(cancellationToken));
+  }
+
+  CompletableFuture<Void> retrieve(
+      Message.Builder builder, CancellationToken<CancellationException> cancellationToken) {
+    return PayloadVisitors.visit(builder, retrieveOptions(cancellationToken));
+  }
+
+  private PayloadVisitorOptions<StorageDriverTargetInfo> storeOptions(
+      @Nullable StorageDriverTargetInfo target,
+      CancellationToken<CancellationException> cancellationToken) {
+    return PayloadVisitorOptions.<StorageDriverTargetInfo>newBuilder(
+            (visitedTarget, payloads) ->
+                payloadTransformer.store(payloads, visitedTarget, cancellationToken))
+        .setInitialContext(target)
+        .setConcurrency(payloadVisitConcurrency)
+        .setSkipSearchAttributes(true)
+        .build();
+  }
+
+  private PayloadVisitorOptions<Void> retrieveOptions(
+      CancellationToken<CancellationException> cancellationToken) {
+    return PayloadVisitorOptions.<Void>newBuilder(
+            (context, payloads) -> payloadTransformer.retrieve(payloads, cancellationToken))
+        .setConcurrency(payloadVisitConcurrency)
+        .setSkipSearchAttributes(true)
+        .build();
+  }
+}
diff --git a/temporal-sdk/src/main/java/io/temporal/internal/payload/storage/ExternalStoragePayloadTransformer.java b/temporal-sdk/src/main/java/io/temporal/internal/payload/storage/ExternalStoragePayloadTransformer.java
new file mode 100644
index 000000000..6e0d4d770
--- /dev/null
+++ b/temporal-sdk/src/main/java/io/temporal/internal/payload/storage/ExternalStoragePayloadTransformer.java
@@ -0,0 +1,292 @@
+package io.temporal.internal.payload.storage;
+
+import io.temporal.api.common.v1.Payload;
+import io.temporal.common.CancellationToken;
+import io.temporal.internal.common.ListUtils;
+import io.temporal.internal.concurrent.structured.TaskScope;
+import io.temporal.payload.storage.ExternalStorageOptions;
+import io.temporal.payload.storage.StorageDriver;
+import io.temporal.payload.storage.StorageDriverClaim;
+import io.temporal.payload.storage.StorageDriverRetrieveContext;
+import io.temporal.payload.storage.StorageDriverSelector;
+import io.temporal.payload.storage.StorageDriverStoreContext;
+import io.temporal.payload.storage.StorageDriverTargetInfo;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Function;
+import javax.annotation.Nullable;
+
+/**
+ * Transforms one payload list between inline payloads and external-storage references by routing
+ * entries to storage drivers.
+ */
+final class ExternalStoragePayloadTransformer {
+  private final Map<String, StorageDriver> driversByName;
+  private final StorageDriverSelector selector;
+  private final int payloadSizeThreshold;
+
+  static ExternalStoragePayloadTransformer fromOptions(ExternalStorageOptions options) {
+    Map<String, StorageDriver> driversByName = new LinkedHashMap<>();
+    for (StorageDriver driver : options.getDrivers()) {
+      driversByName.put(driver.getName(), driver);
+    }
+    return new ExternalStoragePayloadTransformer(
+        driversByName, options.getDriverSelector(), options.getPayloadSizeThreshold());
+  }
+
+  private ExternalStoragePayloadTransformer(
+      Map<String, StorageDriver> driversByName,
+      StorageDriverSelector selector,
+      int payloadSizeThreshold) {
+    this.driversByName = driversByName;
+    this.selector = selector;
+    this.payloadSizeThreshold = payloadSizeThreshold;
+  }
+
+  CompletableFuture<List<Payload>> store(
+      List<Payload> payloads,
+      @Nullable StorageDriverTargetInfo target,
+      CancellationToken<CancellationException> cancellationToken) {
+    StorageDriverStoreContext context =
+        new StorageDriverStoreContextImpl(target, cancellationToken);
+    Map<String, Batch<Payload>> batches;
+    try {
+      batches = buildStoreBatches(payloads, context);
+    } catch (RuntimeException e) {
+      return failedFuture(e);
+    }
+    if (batches.isEmpty()) {
+      return CompletableFuture.completedFuture(payloads);
+    }
+    return runStoreDrivers(batches, target, cancellationToken)
+        .thenApply(referencePayloads -> applyPayloadReplacements(payloads, referencePayloads));
+  }
+
+  private Map<String, Batch<Payload>> buildStoreBatches(
+      List<Payload> payloads, StorageDriverStoreContext context) {
+    Map<String, Batch<Payload>> batches = new LinkedHashMap<>();
+    for (int i = 0; i < payloads.size(); i++) {
+      Payload payload = payloads.get(i);
+      if (payloadSizeThreshold > 0 && payload.getSerializedSize() < payloadSizeThreshold) {
+        continue;
+      }
+      StorageDriver driver = selector.selectDriver(context, payload);
+      if (driver == null) {
+        continue;
+      }
+      if (driversByName.get(driver.getName()) != driver) {
+        throw new IllegalStateException(
+            "Storage driver selector returned a driver not registered with this external storage: '"
+                + driver.getName()
+                + "'");
+      }
+      batches.computeIfAbsent(driver.getName(), name -> new Batch<>(driver)).add(i, payload);
+    }
+    return batches;
+  }
+
+  private CompletableFuture<List<IndexedValue<Payload>>> runStoreDrivers(
+      Map<String, Batch<Payload>> batches,
+      @Nullable StorageDriverTargetInfo target,
+      CancellationToken<CancellationException> cancellationToken) {
+    return withDriverScope(
+        cancellationToken,
+        scope -> {
+          StorageDriverStoreContext context =
+              new StorageDriverStoreContextImpl(target, scope.token());
+          for (Batch<Payload> batch : batches.values()) {
+            scope
+                .attach(batch.driver.store(context, batch.values()))
+                .map(claims -> createReferencePayloads(batch, claims));
+          }
+          return scope.awaitAll(ListUtils::flatten);
+        });
+  }
+
+  /**
+   * Runs {@code body} in a scope that is also cancelled by {@code cancellationToken}, so a caller
+   * abandoning the operation trips the token the drivers observe.
+   */
+  private static CompletableFuture<List<IndexedValue<Payload>>> withDriverScope(
+      CancellationToken<CancellationException> cancellationToken,
+      Function<
+              TaskScope<List<IndexedValue<Payload>>>,
+              CompletableFuture<List<IndexedValue<Payload>>>>
+          body) {
+    return TaskScope.withScope(
+        (TaskScope<List<IndexedValue<Payload>>> scope) -> {
+          CancellationToken.Registration registration =
+              cancellationToken.onCancel(scope::cancelAll);
+          CompletableFuture<List<IndexedValue<Payload>>> result;
+          try {
+            result = body.apply(scope);
+          } catch (Throwable t) {
+            scope.cancelAll();
+            result = failedFuture(t);
+          }
+          // The registration outlives body(), so it is released only once the work settles.
+          return result.whenComplete((ignored, error) -> registration.close());
+        });
+  }
+
+  private static List<IndexedValue<Payload>> createReferencePayloads(
+      Batch<Payload> batch, List<StorageDriverClaim> claims) {
+    if (claims == null || claims.size() != batch.size()) {
+      throw new IllegalStateException(
+          String.format(
+              "Storage driver '%s' returned %d claims for %d payloads",
+              batch.driver.getName(), claims == null ? 0 : claims.size(), batch.size()));
+    }
+    List<IndexedValue<Payload>> replacements = new ArrayList<>(claims.size());
+    for (int batchIndex = 0; batchIndex < claims.size(); batchIndex++) {
+      StorageDriverClaim claim = claims.get(batchIndex);
+      if (claim == null) {
+        throw new IllegalStateException(
+            String.format(
+                "Storage driver '%s' returned a null claim at index %d",
+                batch.driver.getName(), batchIndex));
+      }
+      IndexedValue<Payload> indexedPayload = batch.get(batchIndex);
+      replacements.add(
+          new IndexedValue<>(
+              indexedPayload.originalIndex,
+              ExternalStorageReferences.toReferencePayload(
+                  batch.driver.getName(), claim, indexedPayload.value.getSerializedSize())));
+    }
+    return replacements;
+  }
+
+  CompletableFuture<List<Payload>> retrieve(
+      List<Payload> payloads, CancellationToken<CancellationException> cancellationToken) {
+    Map<String, Batch<StorageDriverClaim>> batches;
+    try {
+      batches = buildRetrieveBatches(payloads);
+    } catch (RuntimeException e) {
+      return failedFuture(e);
+    }
+    if (batches.isEmpty()) {
+      return CompletableFuture.completedFuture(payloads);
+    }
+    return runRetrieveDrivers(batches, cancellationToken)
+        .thenApply(retrievedPayloads -> applyPayloadReplacements(payloads, retrievedPayloads));
+  }
+
+  private Map<String, Batch<StorageDriverClaim>> buildRetrieveBatches(List<Payload> payloads) {
+    Map<String, Batch<StorageDriverClaim>> batches = new LinkedHashMap<>();
+    for (int i = 0; i < payloads.size(); i++) {
+      Payload payload = payloads.get(i);
+      ExternalStorageReferences.ParsedReference reference =
+          ExternalStorageReferences.tryParseReference(payload);
+      if (reference == null) {
+        continue;
+      }
+      StorageDriver driver = driversByName.get(reference.driverName);
+      if (driver == null) {
+        throw new IllegalStateException(
+            "No storage driver registered with name '" + reference.driverName + "'");
+      }
+      batches
+          .computeIfAbsent(reference.driverName, name -> new Batch<>(driver))
+          .add(i, reference.claim);
+    }
+    return batches;
+  }
+
+  private CompletableFuture<List<IndexedValue<Payload>>> runRetrieveDrivers(
+      Map<String, Batch<StorageDriverClaim>> batches,
+      CancellationToken<CancellationException> cancellationToken) {
+    return withDriverScope(
+        cancellationToken,
+        scope -> {
+          StorageDriverRetrieveContext context =
+              new StorageDriverRetrieveContextImpl(scope.token());
+          for (Batch<StorageDriverClaim> batch : batches.values()) {
+            scope
+                .attach(batch.driver.retrieve(context, batch.values()))
+                .map(payloads -> mapPayloadsToOriginalPositions(batch, payloads));
+          }
+          return scope.awaitAll(ListUtils::flatten);
+        });
+  }
+
+  private static List<IndexedValue<Payload>> mapPayloadsToOriginalPositions(
+      Batch<StorageDriverClaim> batch, List<Payload> payloads) {
+    if (payloads == null || payloads.size() != batch.size()) {
+      throw new IllegalStateException(
+          String.format(
+              "Storage driver '%s' returned %d payloads for %d claims",
+              batch.driver.getName(), payloads == null ? 0 : payloads.size(), batch.size()));
+    }
+    List<IndexedValue<Payload>> replacements = new ArrayList<>(payloads.size());
+    for (int batchIndex = 0; batchIndex < payloads.size(); batchIndex++) {
+      Payload payload = payloads.get(batchIndex);
+      if (payload == null) {
+        throw new IllegalStateException(
+            String.format(
+                "Storage driver '%s' returned a null payload at index %d",
+                batch.driver.getName(), batchIndex));
+      }
+      replacements.add(new IndexedValue<>(batch.get(batchIndex).originalIndex, payload));
+    }
+    return replacements;
+  }
+
+  private static <T> CompletableFuture<T> failedFuture(Throwable t) {
+    CompletableFuture<T> future = new CompletableFuture<>();
+    future.completeExceptionally(t);
+    return future;
+  }
+
+  private static List<Payload> applyPayloadReplacements(
+      List<Payload> payloads, List<IndexedValue<Payload>> replacements) {
+    Payload[] updatedPayloads = payloads.toArray(new Payload[0]);
+    for (IndexedValue<Payload> replacement : replacements) {
+      updatedPayloads[replacement.originalIndex] = replacement.value;
+    }
+    return Arrays.asList(updatedPayloads);
+  }
+
+  private static final class IndexedValue<T> {
+    final int originalIndex;
+    final T value;
+
+    IndexedValue(int originalIndex, T value) {
+      this.originalIndex = originalIndex;
+      this.value = value;
+    }
+  }
+
+  private static final class Batch<T> {
+    final StorageDriver driver;
+    private final List<IndexedValue<T>> indexedValues = new ArrayList<>();
+
+    Batch(StorageDriver driver) {
+      this.driver = driver;
+    }
+
+    void add(int originalIndex, T value) {
+      indexedValues.add(new IndexedValue<>(originalIndex, value));
+    }
+
+    int size() {
+      return indexedValues.size();
+    }
+
+    IndexedValue<T> get(int batchIndex) {
+      return indexedValues.get(batchIndex);
+    }
+
+    List<T> values() {
+      List<T> values = new ArrayList<>(indexedValues.size());
+      for (IndexedValue<T> indexedValue : indexedValues) {
+        values.add(indexedValue.value);
+      }
+      return values;
+    }
+  }
+}
diff --git a/temporal-sdk/src/main/java/io/temporal/internal/payload/storage/ExternalStorageReferences.java b/temporal-sdk/src/main/java/io/temporal/internal/payload/storage/ExternalStorageReferences.java
new file mode 100644
index 000000000..3a68c6bb6
--- /dev/null
+++ b/temporal-sdk/src/main/java/io/temporal/internal/payload/storage/ExternalStorageReferences.java
@@ -0,0 +1,88 @@
+package io.temporal.internal.payload.storage;
+
+import com.google.protobuf.ByteString;
+import com.google.protobuf.util.JsonFormat;
+import io.temporal.api.common.v1.Payload;
+import io.temporal.api.sdk.v1.ExternalStorageReference;
+import io.temporal.common.converter.EncodingKeys;
+import io.temporal.payload.storage.StorageDriverClaim;
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+final class ExternalStorageReferences {
+  private static final String ENCODING_PROTOBUF_JSON = "json/protobuf";
+  private static final String REFERENCE_MESSAGE_TYPE =
+      ExternalStorageReference.getDescriptor().getFullName();
+
+  private static final JsonFormat.Printer PRINTER = JsonFormat.printer();
+  private static final JsonFormat.Parser PARSER = JsonFormat.parser().ignoringUnknownFields();
+
+  static final class ParsedReference {
+    final String driverName;
+    final StorageDriverClaim claim;
+
+    ParsedReference(String driverName, StorageDriverClaim claim) {
+      this.driverName = driverName;
+      this.claim = claim;
+    }
+  }
+
+  static Payload toReferencePayload(
+      @Nonnull String driverName,
+      @Nonnull StorageDriverClaim claim,
+      long originalPayloadSizeBytes) {
+    ExternalStorageReference reference =
+        ExternalStorageReference.newBuilder()
+            .setDriverName(driverName)
+            .putAllClaimData(claim.getClaimData())
+            .build();
+    String json;
+    try {
+      json = PRINTER.print(reference);
+    } catch (Exception e) {
+      throw new IllegalStateException("Failed to serialize external storage reference", e);
+    }
+    return Payload.newBuilder()
+        .putMetadata(
+            EncodingKeys.METADATA_ENCODING_KEY, ByteString.copyFromUtf8(ENCODING_PROTOBUF_JSON))
+        .putMetadata(
+            EncodingKeys.METADATA_MESSAGE_TYPE_KEY, ByteString.copyFromUtf8(REFERENCE_MESSAGE_TYPE))
+        .setData(ByteString.copyFromUtf8(json))
+        .addExternalPayloads(
+            Payload.ExternalPayloadDetails.newBuilder()
+                .setSizeBytes(originalPayloadSizeBytes)
+                .build())
+        .build();
+  }
+
+  /**
+   * Returns the reference encoded in {@code payload}, or null if the payload is not an external
+   * storage reference this SDK understands.
+   *
+   * <p>Only the encoding and message type identify a reference. {@code external_payloads} records
+   * the original size for the server's benefit and is not part of the exchange contract, so a
+   * producer that omits it still yields a readable reference.
+   */
+  static @Nullable ParsedReference tryParseReference(@Nonnull Payload payload) {
+    if (!hasMetadata(payload, EncodingKeys.METADATA_ENCODING_KEY, ENCODING_PROTOBUF_JSON)
+        || !hasMetadata(payload, EncodingKeys.METADATA_MESSAGE_TYPE_KEY, REFERENCE_MESSAGE_TYPE)) {
+      return null;
+    }
+    ExternalStorageReference.Builder builder = ExternalStorageReference.newBuilder();
+    try {
+      PARSER.merge(payload.getData().toStringUtf8(), builder);
+    } catch (Exception e) {
+      throw new IllegalArgumentException("Failed to parse external storage reference", e);
+    }
+    ExternalStorageReference reference = builder.build();
+    return new ParsedReference(
+        reference.getDriverName(), new StorageDriverClaim(reference.getClaimDataMap()));
+  }
+
+  private static boolean hasMetadata(Payload payload, String key, String expected) {
+    ByteString value = payload.getMetadataMap().get(key);
+    return value != null && expected.equals(value.toStringUtf8());
+  }
+
+  private ExternalStorageReferences() {}
+}
diff --git a/temporal-sdk/src/main/java/io/temporal/internal/payload/storage/StorageDriverRetrieveContextImpl.java b/temporal-sdk/src/main/java/io/temporal/internal/payload/storage/StorageDriverRetrieveContextImpl.java
new file mode 100644
index 000000000..84374b2cd
--- /dev/null
+++ b/temporal-sdk/src/main/java/io/temporal/internal/payload/storage/StorageDriverRetrieveContextImpl.java
@@ -0,0 +1,21 @@
+package io.temporal.internal.payload.storage;
+
+import io.temporal.common.CancellationToken;
+import io.temporal.payload.storage.StorageDriverRetrieveContext;
+import java.util.Objects;
+import java.util.concurrent.CancellationException;
+import javax.annotation.Nonnull;
+
+final class StorageDriverRetrieveContextImpl implements StorageDriverRetrieveContext {
+  private final CancellationToken<CancellationException> cancellationToken;
+
+  StorageDriverRetrieveContextImpl(CancellationToken<CancellationException> cancellationToken) {
+    this.cancellationToken = Objects.requireNonNull(cancellationToken, "cancellationToken");
+  }
+
+  @Nonnull
+  @Override
+  public CancellationToken<CancellationException> getCancellationToken() {
+    return cancellationToken;
+  }
+}
diff --git a/temporal-sdk/src/main/java/io/temporal/internal/payload/storage/StorageDriverStoreContextImpl.java b/temporal-sdk/src/main/java/io/temporal/internal/payload/storage/StorageDriverStoreContextImpl.java
new file mode 100644
index 000000000..c28ea0f63
--- /dev/null
+++ b/temporal-sdk/src/main/java/io/temporal/internal/payload/storage/StorageDriverStoreContextImpl.java
@@ -0,0 +1,33 @@
+package io.temporal.internal.payload.storage;
+
+import io.temporal.common.CancellationToken;
+import io.temporal.payload.storage.StorageDriverStoreContext;
+import io.temporal.payload.storage.StorageDriverTargetInfo;
+import java.util.Objects;
+import java.util.concurrent.CancellationException;
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+final class StorageDriverStoreContextImpl implements StorageDriverStoreContext {
+  private final @Nullable StorageDriverTargetInfo target;
+  private final CancellationToken<CancellationException> cancellationToken;
+
+  StorageDriverStoreContextImpl(
+      @Nullable StorageDriverTargetInfo target,
+      CancellationToken<CancellationException> cancellationToken) {
+    this.target = target;
+    this.cancellationToken = Objects.requireNonNull(cancellationToken, "cancellationToken");
+  }
+
+  @Nullable
+  @Override
+  public StorageDriverTargetInfo getTarget() {
+    return target;
+  }
+
+  @Nonnull
+  @Override
+  public CancellationToken<CancellationException> getCancellationToken() {
+    return cancellationToken;
+  }
+}
diff --git a/temporal-sdk/src/main/java/io/temporal/internal/payload/visitor/PayloadVisitor.java b/temporal-sdk/src/main/java/io/temporal/internal/payload/visitor/PayloadVisitor.java
index 8872a44cd..70a2c1cc4 100644
--- a/temporal-sdk/src/main/java/io/temporal/internal/payload/visitor/PayloadVisitor.java
+++ b/temporal-sdk/src/main/java/io/temporal/internal/payload/visitor/PayloadVisitor.java
@@ -17,6 +17,6 @@
  * @param <C> type of the contextual value supplied to each visit
  */
 @FunctionalInterface
-interface PayloadVisitor<C> {
+public interface PayloadVisitor<C> {
   CompletableFuture<List<Payload>> visit(C context, List<Payload> payloads);
 }
diff --git a/temporal-sdk/src/main/java/io/temporal/internal/payload/visitor/PayloadVisitorOptions.java b/temporal-sdk/src/main/java/io/temporal/internal/payload/visitor/PayloadVisitorOptions.java
index c834c2ec8..4eac39be4 100644
--- a/temporal-sdk/src/main/java/io/temporal/internal/payload/visitor/PayloadVisitorOptions.java
+++ b/temporal-sdk/src/main/java/io/temporal/internal/payload/visitor/PayloadVisitorOptions.java
@@ -9,7 +9,7 @@
  *
  * @param <C> type of the contextual value supplied to the visitor
  */
-final class PayloadVisitorOptions<C> {
+public final class PayloadVisitorOptions<C> {
   private final @Nonnull PayloadVisitor<C> payloadVisitor;
   private final @Nullable MessageVisitor<C> messageVisitor;
   private final @Nullable C initialContext;
@@ -36,7 +36,7 @@ public PayloadVisitor<C> getPayloadVisitor() {
   }
 
   @Nullable
-  public MessageVisitor<C> getMessageVisitor() {
+  MessageVisitor<C> getMessageVisitor() {
     return messageVisitor;
   }
 
@@ -69,7 +69,7 @@ private Builder(@Nonnull PayloadVisitor<C> payloadVisitor) {
       this.payloadVisitor = Objects.requireNonNull(payloadVisitor, "payloadVisitor");
     }
 
-    public Builder<C> setMessageVisitor(@Nullable MessageVisitor<C> messageVisitor) {
+    Builder<C> setMessageVisitor(@Nullable MessageVisitor<C> messageVisitor) {
       this.messageVisitor = messageVisitor;
       return this;
     }
diff --git a/temporal-sdk/src/main/java/io/temporal/internal/payload/visitor/PayloadVisitors.java b/temporal-sdk/src/main/java/io/temporal/internal/payload/visitor/PayloadVisitors.java
index 69e992412..c6f6179fa 100644
--- a/temporal-sdk/src/main/java/io/temporal/internal/payload/visitor/PayloadVisitors.java
+++ b/temporal-sdk/src/main/java/io/temporal/internal/payload/visitor/PayloadVisitors.java
@@ -5,7 +5,7 @@
 import javax.annotation.Nonnull;
 
 /** Visits every payload within a proto message. */
-final class PayloadVisitors {
+public final class PayloadVisitors {
   private PayloadVisitors() {}
 
   /** Visits the payloads in {@code builder} in place. */
diff --git a/temporal-sdk/src/main/java/io/temporal/payload/storage/StorageDriver.java b/temporal-sdk/src/main/java/io/temporal/payload/storage/StorageDriver.java
index 323925075..01d851fbe 100644
--- a/temporal-sdk/src/main/java/io/temporal/payload/storage/StorageDriver.java
+++ b/temporal-sdk/src/main/java/io/temporal/payload/storage/StorageDriver.java
@@ -29,6 +29,9 @@ public interface StorageDriver {
   /**
    * Stores {@code payloads} and returns one {@link StorageDriverClaim} per payload, in the same
    * order. The returned list must be the same length as {@code payloads}.
+   *
+   * <p>Observe {@link StorageDriverStoreContext#getCancellationToken()} to learn when the SDK has
+   * abandoned this operation, and abandon in-flight requests accordingly.
    */
   @Nonnull
   CompletableFuture<List<StorageDriverClaim>> store(
@@ -37,6 +40,9 @@ CompletableFuture<List<StorageDriverClaim>> store(
   /**
    * Retrieves the payloads identified by {@code claims} and returns one {@link Payload} per claim,
    * in the same order. The returned list must be the same length as {@code claims}.
+   *
+   * <p>Observe {@link StorageDriverRetrieveContext#getCancellationToken()} to learn when the SDK
+   * has abandoned this operation, and abandon in-flight requests accordingly.
    */
   @Nonnull
   CompletableFuture<List<Payload>> retrieve(
diff --git a/temporal-sdk/src/main/java/io/temporal/payload/storage/StorageDriverRetrieveContext.java b/temporal-sdk/src/main/java/io/temporal/payload/storage/StorageDriverRetrieveContext.java
index 77f11c750..ae2b1ffbe 100644
--- a/temporal-sdk/src/main/java/io/temporal/payload/storage/StorageDriverRetrieveContext.java
+++ b/temporal-sdk/src/main/java/io/temporal/payload/storage/StorageDriverRetrieveContext.java
@@ -1,12 +1,25 @@
 package io.temporal.payload.storage;
 
+import io.temporal.common.CancellationToken;
 import io.temporal.common.Experimental;
+import java.util.concurrent.CancellationException;
+import javax.annotation.Nonnull;
 
 /**
  * Context passed to {@link StorageDriver#retrieve}.
  *
- * <p>Implemented by the SDK and passed to the driver. Driver authors do not implement this in
- * production code, only when constructing instances for their own tests.
+ * <p>The SDK supplies the instance a driver receives. Members added here in later releases will
+ * carry a default, so an existing driver-side implementation keeps compiling and behaves as though
+ * the new member were absent.
  */
 @Experimental
-public interface StorageDriverRetrieveContext {}
+public interface StorageDriverRetrieveContext {
+  /**
+   * Token cancelled when the SDK abandons this retrieve operation. Defaults to a token that is
+   * never cancelled.
+   */
+  @Nonnull
+  default CancellationToken<CancellationException> getCancellationToken() {
+    return CancellationToken.none();
+  }
+}
diff --git a/temporal-sdk/src/main/java/io/temporal/payload/storage/StorageDriverStoreContext.java b/temporal-sdk/src/main/java/io/temporal/payload/storage/StorageDriverStoreContext.java
index 723655bf1..f001adc57 100644
--- a/temporal-sdk/src/main/java/io/temporal/payload/storage/StorageDriverStoreContext.java
+++ b/temporal-sdk/src/main/java/io/temporal/payload/storage/StorageDriverStoreContext.java
@@ -1,13 +1,17 @@
 package io.temporal.payload.storage;
 
+import io.temporal.common.CancellationToken;
 import io.temporal.common.Experimental;
+import java.util.concurrent.CancellationException;
+import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
 
 /**
  * Context passed to {@link StorageDriver#store} and {@link StorageDriverSelector}.
  *
- * <p>Implemented by the SDK and passed to the driver. Driver authors do not implement this in
- * production code, only when constructing instances for their own tests.
+ * <p>The SDK supplies the instance a driver receives. Members added here in later releases will
+ * carry a default, so an existing driver-side implementation keeps compiling and behaves as though
+ * the new member were absent.
  */
 @Experimental
 public interface StorageDriverStoreContext {
@@ -17,4 +21,13 @@ public interface StorageDriverStoreContext {
    */
   @Nullable
   StorageDriverTargetInfo getTarget();
+
+  /**
+   * Token cancelled when the SDK abandons this store operation. Defaults to a token that is never
+   * cancelled.
+   */
+  @Nonnull
+  default CancellationToken<CancellationException> getCancellationToken() {
+    return CancellationToken.none();
+  }
 }
diff --git a/temporal-sdk/src/test/java/io/temporal/internal/payload/storage/ExternalStorageMessageTransformerTest.java b/temporal-sdk/src/test/java/io/temporal/internal/payload/storage/ExternalStorageMessageTransformerTest.java
new file mode 100644
index 000000000..f17bcff47
--- /dev/null
+++ b/temporal-sdk/src/test/java/io/temporal/internal/payload/storage/ExternalStorageMessageTransformerTest.java
@@ -0,0 +1,161 @@
+package io.temporal.internal.payload.storage;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+import com.google.protobuf.ByteString;
+import io.temporal.api.command.v1.Command;
+import io.temporal.api.command.v1.ScheduleActivityTaskCommandAttributes;
+import io.temporal.api.command.v1.StartChildWorkflowExecutionCommandAttributes;
+import io.temporal.api.common.v1.Payload;
+import io.temporal.api.common.v1.Payloads;
+import io.temporal.api.common.v1.SearchAttributes;
+import io.temporal.common.CancellationToken;
+import io.temporal.payload.storage.ExternalStorageOptions;
+import io.temporal.payload.storage.StorageDriver;
+import io.temporal.payload.storage.StorageDriverClaim;
+import io.temporal.payload.storage.StorageDriverRetrieveContext;
+import io.temporal.payload.storage.StorageDriverStoreContext;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import org.junit.Test;
+
+/** Tests external storage message conversion. */
+public class ExternalStorageMessageTransformerTest {
+
+  @Test
+  public void storeAndRetrieveRoundTripsOverAMessage() throws Exception {
+    InMemoryDriver driver = new InMemoryDriver("d1");
+    ExternalStorageMessageTransformer transformer = transformer(driver, 0);
+    Payloads message =
+        Payloads.newBuilder().addPayloads(payload("a")).addPayloads(payload("b")).build();
+
+    Payloads stored = transformer.store(message, null, CancellationToken.none()).get();
+
+    assertNotNull(ExternalStorageReferences.tryParseReference(stored.getPayloads(0)));
+    assertNotNull(ExternalStorageReferences.tryParseReference(stored.getPayloads(1)));
+
+    Payloads retrieved = transformer.retrieve(stored, CancellationToken.none()).get();
+    assertEquals(message, retrieved);
+  }
+
+  @Test
+  public void walksNestedPayloads() throws Exception {
+    InMemoryDriver driver = new InMemoryDriver("d1");
+    ExternalStorageMessageTransformer transformer = transformer(driver, 0);
+    Command command =
+        Command.newBuilder()
+            .setScheduleActivityTaskCommandAttributes(
+                ScheduleActivityTaskCommandAttributes.newBuilder()
+                    .setInput(Payloads.newBuilder().addPayloads(payload("deep"))))
+            .build();
+
+    Command stored = transformer.store(command, null, CancellationToken.none()).get();
+
+    Payload nested = stored.getScheduleActivityTaskCommandAttributes().getInput().getPayloads(0);
+    assertNotNull(ExternalStorageReferences.tryParseReference(nested));
+    assertEquals(command, transformer.retrieve(stored, CancellationToken.none()).get());
+  }
+
+  @Test
+  public void payloadBelowThresholdLeavesMessageUnchanged() throws Exception {
+    InMemoryDriver driver = new InMemoryDriver("d1");
+    ExternalStorageMessageTransformer transformer = transformer(driver, 1024);
+    Payloads message = Payloads.newBuilder().addPayloads(payload("small")).build();
+
+    Payloads stored = transformer.store(message, null, CancellationToken.none()).get();
+
+    assertNull(ExternalStorageReferences.tryParseReference(stored.getPayloads(0)));
+    assertEquals(message, stored);
+    assertTrue(driver.storeBatchSizes.isEmpty());
+  }
+
+  @Test
+  public void searchAttributesAreNotOffloaded() throws Exception {
+    InMemoryDriver driver = new InMemoryDriver("d1");
+    ExternalStorageMessageTransformer transformer = transformer(driver, 0);
+    Command command =
+        Command.newBuilder()
+            .setStartChildWorkflowExecutionCommandAttributes(
+                StartChildWorkflowExecutionCommandAttributes.newBuilder()
+                    .setInput(Payloads.newBuilder().addPayloads(payload("input")))
+                    .setSearchAttributes(
+                        SearchAttributes.newBuilder()
+                            .putIndexedFields("k", payload("indexed-value"))))
+            .build();
+
+    Command stored = transformer.store(command, null, CancellationToken.none()).get();
+
+    StartChildWorkflowExecutionCommandAttributes attrs =
+        stored.getStartChildWorkflowExecutionCommandAttributes();
+    assertNotNull(ExternalStorageReferences.tryParseReference(attrs.getInput().getPayloads(0)));
+    Payload indexed = attrs.getSearchAttributes().getIndexedFieldsOrThrow("k");
+    assertNull(ExternalStorageReferences.tryParseReference(indexed));
+    assertEquals(payload("indexed-value"), indexed);
+  }
+
+  private static ExternalStorageMessageTransformer transformer(
+      StorageDriver driver, int threshold) {
+    ExternalStoragePayloadTransformer payloadTransformer =
+        ExternalStoragePayloadTransformer.fromOptions(
+            ExternalStorageOptions.newBuilder()
+                .setDriver(driver)
+                .setPayloadSizeThreshold(threshold)
+                .build());
+    return new ExternalStorageMessageTransformer(payloadTransformer, 4);
+  }
+
+  private static Payload payload(String data) {
+    return Payload.newBuilder().setData(ByteString.copyFromUtf8(data)).build();
+  }
+
+  private static final class InMemoryDriver implements StorageDriver {
+    private final String name;
+    private final Map<String, Payload> objects = new HashMap<>();
+    final List<Integer> storeBatchSizes = new ArrayList<>();
+    private int counter = 0;
+
+    InMemoryDriver(String name) {
+      this.name = name;
+    }
+
+    @Override
+    public String getName() {
+      return name;
+    }
+
+    @Override
+    public String getType() {
+      return "test.inmemory";
+    }
+
+    @Override
+    public synchronized CompletableFuture<List<StorageDriverClaim>> store(
+        StorageDriverStoreContext context, List<Payload> payloads) {
+      storeBatchSizes.add(payloads.size());
+      List<StorageDriverClaim> claims = new ArrayList<>();
+      for (Payload payload : payloads) {
+        String key = name + "-" + (counter++);
+        objects.put(key, payload);
+        claims.add(new StorageDriverClaim(Collections.singletonMap("key", key)));
+      }
+      return CompletableFuture.completedFuture(claims);
+    }
+
+    @Override
+    public synchronized CompletableFuture<List<Payload>> retrieve(
+        StorageDriverRetrieveContext context, List<StorageDriverClaim> claims) {
+      List<Payload> payloads = new ArrayList<>();
+      for (StorageDriverClaim claim : claims) {
+        payloads.add(objects.get(claim.getClaimData().get("key")));
+      }
+      return CompletableFuture.completedFuture(payloads);
+    }
+  }
+}
diff --git a/temporal-sdk/src/test/java/io/temporal/internal/payload/storage/ExternalStoragePayloadTransformerTest.java b/temporal-sdk/src/test/java/io/temporal/internal/payload/storage/ExternalStoragePayloadTransformerTest.java
new file mode 100644
index 000000000..f1632ca81
--- /dev/null
+++ b/temporal-sdk/src/test/java/io/temporal/internal/payload/storage/ExternalStoragePayloadTransformerTest.java
@@ -0,0 +1,441 @@
+package io.temporal.internal.payload.storage;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import com.google.protobuf.ByteString;
+import io.temporal.api.common.v1.Payload;
+import io.temporal.common.CancellationToken;
+import io.temporal.internal.concurrent.structured.CancelSource;
+import io.temporal.payload.storage.ExternalStorageOptions;
+import io.temporal.payload.storage.StorageDriver;
+import io.temporal.payload.storage.StorageDriverClaim;
+import io.temporal.payload.storage.StorageDriverRetrieveContext;
+import io.temporal.payload.storage.StorageDriverSelector;
+import io.temporal.payload.storage.StorageDriverStoreContext;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+import org.junit.Test;
+
+/** Tests external storage payload-list conversion. */
+public class ExternalStoragePayloadTransformerTest {
+
+  @Test
+  public void storesAndRetrievesRoundTrip() throws Exception {
+    InMemoryDriver driver = new InMemoryDriver("d1");
+    ExternalStoragePayloadTransformer transformer = transformer(driver, 0);
+    List<Payload> input = Arrays.asList(payload("a"), payload("b"));
+
+    List<Payload> stored = transformer.store(input, null, CancellationToken.none()).get();
+
+    assertEquals(2, stored.size());
+    assertNotNull(ExternalStorageReferences.tryParseReference(stored.get(0)));
+    assertNotNull(ExternalStorageReferences.tryParseReference(stored.get(1)));
+    assertEquals(Collections.singletonList(2), driver.storeBatchSizes);
+    assertEquals(
+        input.get(0).getSerializedSize(), stored.get(0).getExternalPayloads(0).getSizeBytes());
+
+    List<Payload> retrieved = transformer.retrieve(stored, CancellationToken.none()).get();
+    assertEquals(input, retrieved);
+    assertEquals(Collections.singletonList(2), driver.retrieveBatchSizes);
+  }
+
+  @Test
+  public void payloadBelowThresholdStaysInline() throws Exception {
+    InMemoryDriver driver = new InMemoryDriver("d1");
+    ExternalStoragePayloadTransformer transformer = transformer(driver, 100);
+    Payload small = payload("x");
+    Payload large = payload(repeat("y", 200));
+
+    List<Payload> stored =
+        transformer.store(Arrays.asList(small, large), null, CancellationToken.none()).get();
+
+    assertNull(ExternalStorageReferences.tryParseReference(stored.get(0)));
+    assertEquals(small, stored.get(0));
+    assertNotNull(ExternalStorageReferences.tryParseReference(stored.get(1)));
+    assertEquals(Collections.singletonList(1), driver.storeBatchSizes);
+  }
+
+  @Test
+  public void selectorReturningNullKeepsInline() throws Exception {
+    InMemoryDriver driver = new InMemoryDriver("d1");
+    ExternalStoragePayloadTransformer transformer =
+        ExternalStoragePayloadTransformer.fromOptions(
+            ExternalStorageOptions.newBuilder()
+                .setDriver(driver)
+                .setDriverSelector((context, payload) -> null)
+                .setPayloadSizeThreshold(0)
+                .build());
+
+    List<Payload> stored =
+        transformer
+            .store(Collections.singletonList(payload("a")), null, CancellationToken.none())
+            .get();
+
+    assertEquals(payload("a"), stored.get(0));
+    assertTrue(driver.storeBatchSizes.isEmpty());
+  }
+
+  @Test
+  public void multipleDriversBatchPerDriverAndPreserveOrder() throws Exception {
+    InMemoryDriver d1 = new InMemoryDriver("d1");
+    InMemoryDriver d2 = new InMemoryDriver("d2");
+    Map<String, StorageDriver> byPrefix = new HashMap<>();
+    byPrefix.put("1", d1);
+    byPrefix.put("2", d2);
+    StorageDriverSelector selector =
+        (context, payload) -> byPrefix.get(payload.getData().toStringUtf8().substring(0, 1));
+    ExternalStoragePayloadTransformer transformer =
+        ExternalStoragePayloadTransformer.fromOptions(
+            ExternalStorageOptions.newBuilder()
+                .setDrivers(Arrays.asList(d1, d2))
+                .setDriverSelector(selector)
+                .setPayloadSizeThreshold(0)
+                .build());
+    List<Payload> input = Arrays.asList(payload("1-a"), payload("2-b"), payload("1-c"));
+
+    List<Payload> stored = transformer.store(input, null, CancellationToken.none()).get();
+
+    assertEquals(Collections.singletonList(2), d1.storeBatchSizes);
+    assertEquals(Collections.singletonList(1), d2.storeBatchSizes);
+    assertEquals(input, transformer.retrieve(stored, CancellationToken.none()).get());
+  }
+
+  @Test
+  public void arityMismatchFails() {
+    StorageDriver driver =
+        new FakeDriver("d1") {
+          @Override
+          public CompletableFuture<List<StorageDriverClaim>> store(
+              StorageDriverStoreContext context, List<Payload> payloads) {
+            return CompletableFuture.completedFuture(Collections.emptyList());
+          }
+        };
+    ExternalStoragePayloadTransformer transformer = transformer(driver, 0);
+
+    Throwable cause =
+        causeOf(
+            transformer.store(
+                Collections.singletonList(payload("a")), null, CancellationToken.none()));
+    assertTrue(cause instanceof IllegalStateException);
+    assertTrue(cause.getMessage().contains("returned 0 claims for 1 payloads"));
+  }
+
+  @Test
+  public void nullClaimFromDriverFails() {
+    StorageDriver driver =
+        new FakeDriver("d1") {
+          @Override
+          public CompletableFuture<List<StorageDriverClaim>> store(
+              StorageDriverStoreContext context, List<Payload> payloads) {
+            return CompletableFuture.completedFuture(Collections.singletonList(null));
+          }
+        };
+    ExternalStoragePayloadTransformer transformer = transformer(driver, 0);
+
+    Throwable cause =
+        causeOf(
+            transformer.store(
+                Collections.singletonList(payload("a")), null, CancellationToken.none()));
+    assertTrue(cause instanceof IllegalStateException);
+    assertTrue(cause.getMessage().contains("returned a null claim at index 0"));
+  }
+
+  @Test
+  public void nullPayloadFromDriverFails() {
+    StorageDriver driver =
+        new FakeDriver("d1") {
+          @Override
+          public CompletableFuture<List<Payload>> retrieve(
+              StorageDriverRetrieveContext context, List<StorageDriverClaim> claims) {
+            return CompletableFuture.completedFuture(Collections.singletonList(null));
+          }
+        };
+    ExternalStoragePayloadTransformer transformer = transformer(driver, 0);
+    Payload reference =
+        ExternalStorageReferences.toReferencePayload(
+            "d1", new StorageDriverClaim(Collections.singletonMap("key", "k")), 1L);
+
+    Throwable cause =
+        causeOf(
+            transformer.retrieve(Collections.singletonList(reference), CancellationToken.none()));
+    assertTrue(cause instanceof IllegalStateException);
+    assertTrue(cause.getMessage().contains("returned a null payload at index 0"));
+  }
+
+  @Test
+  public void unknownDriverOnRetrieveFails() {
+    InMemoryDriver driver = new InMemoryDriver("d1");
+    ExternalStoragePayloadTransformer transformer = transformer(driver, 0);
+    Payload reference =
+        ExternalStorageReferences.toReferencePayload(
+            "ghost", new StorageDriverClaim(Collections.singletonMap("key", "k")), 1L);
+
+    Throwable cause =
+        causeOf(
+            transformer.retrieve(Collections.singletonList(reference), CancellationToken.none()));
+    assertTrue(cause instanceof IllegalStateException);
+    assertTrue(cause.getMessage().contains("No storage driver registered with name 'ghost'"));
+  }
+
+  @Test
+  public void selectorReturningUnregisteredDriverFails() {
+    InMemoryDriver registered = new InMemoryDriver("d1");
+    InMemoryDriver stranger = new InMemoryDriver("d2");
+    ExternalStoragePayloadTransformer transformer =
+        ExternalStoragePayloadTransformer.fromOptions(
+            ExternalStorageOptions.newBuilder()
+                .setDriver(registered)
+                .setDriverSelector((context, payload) -> stranger)
+                .setPayloadSizeThreshold(0)
+                .build());
+
+    Throwable cause =
+        causeOf(
+            transformer.store(
+                Collections.singletonList(payload("a")), null, CancellationToken.none()));
+    assertTrue(cause instanceof IllegalStateException);
+    assertTrue(cause.getMessage().contains("not registered"));
+  }
+
+  @Test
+  public void firstErrorRequestsCancellationOfOutstandingDriverCalls() {
+    CompletableFuture<List<StorageDriverClaim>> inFlight = new CompletableFuture<>();
+    CompletableFuture<List<StorageDriverClaim>> failing = new CompletableFuture<>();
+    AtomicBoolean cancellationRequested = new AtomicBoolean(false);
+    StorageDriver slow =
+        new FakeDriver("d1") {
+          @Override
+          public CompletableFuture<List<StorageDriverClaim>> store(
+              StorageDriverStoreContext context, List<Payload> payloads) {
+            context.getCancellationToken().onCancel(() -> cancellationRequested.set(true));
+            return inFlight;
+          }
+        };
+    StorageDriver doomed = controlledStore("d2", failing);
+    Map<String, StorageDriver> byPrefix = new HashMap<>();
+    byPrefix.put("1", slow);
+    byPrefix.put("2", doomed);
+    ExternalStoragePayloadTransformer transformer =
+        ExternalStoragePayloadTransformer.fromOptions(
+            ExternalStorageOptions.newBuilder()
+                .setDrivers(Arrays.asList(slow, doomed))
+                .setDriverSelector(
+                    (context, payload) ->
+                        byPrefix.get(payload.getData().toStringUtf8().substring(0, 1)))
+                .setPayloadSizeThreshold(0)
+                .build());
+
+    CompletableFuture<List<Payload>> result =
+        transformer.store(
+            Arrays.asList(payload("1-a"), payload("2-b")), null, CancellationToken.none());
+    assertFalse(result.isDone());
+
+    failing.completeExceptionally(new RuntimeException("boom"));
+
+    assertTrue(result.isCompletedExceptionally());
+    assertTrue(cancellationRequested.get());
+    assertTrue(inFlight.isCancelled());
+  }
+
+  @Test
+  public void callerCancellationRequestsCancellationOfInFlightStore() {
+    CompletableFuture<List<StorageDriverClaim>> inFlight = new CompletableFuture<>();
+    AtomicBoolean cancellationRequested = new AtomicBoolean(false);
+    StorageDriver slow =
+        new FakeDriver("d1") {
+          @Override
+          public CompletableFuture<List<StorageDriverClaim>> store(
+              StorageDriverStoreContext context, List<Payload> payloads) {
+            context.getCancellationToken().onCancel(() -> cancellationRequested.set(true));
+            return inFlight;
+          }
+        };
+    CancelSource<CancellationException> caller = new CancelSource<>(CancellationException::new);
+
+    CompletableFuture<List<Payload>> result =
+        transformer(slow, 0).store(Collections.singletonList(payload("a")), null, caller.token());
+    assertFalse(result.isDone());
+
+    caller.cancel();
+
+    assertTrue(cancellationRequested.get());
+    assertTrue(inFlight.isCancelled());
+    assertTrue(result.isCompletedExceptionally());
+  }
+
+  @Test
+  public void callerCancellationRequestsCancellationOfInFlightRetrieve() {
+    CompletableFuture<List<Payload>> inFlight = new CompletableFuture<>();
+    AtomicBoolean cancellationRequested = new AtomicBoolean(false);
+    StorageDriver slow =
+        new FakeDriver("d1") {
+          @Override
+          public CompletableFuture<List<Payload>> retrieve(
+              StorageDriverRetrieveContext context, List<StorageDriverClaim> claims) {
+            context.getCancellationToken().onCancel(() -> cancellationRequested.set(true));
+            return inFlight;
+          }
+        };
+    CancelSource<CancellationException> caller = new CancelSource<>(CancellationException::new);
+    Payload reference =
+        ExternalStorageReferences.toReferencePayload(
+            "d1", new StorageDriverClaim(Collections.singletonMap("key", "k")), 1L);
+
+    CompletableFuture<List<Payload>> result =
+        transformer(slow, 0).retrieve(Collections.singletonList(reference), caller.token());
+    assertFalse(result.isDone());
+
+    caller.cancel();
+
+    assertTrue(cancellationRequested.get());
+    assertTrue(inFlight.isCancelled());
+    assertTrue(result.isCompletedExceptionally());
+  }
+
+  @Test
+  public void selectorObservesCallerCancellationToken() {
+    InMemoryDriver driver = new InMemoryDriver("d1");
+    CancelSource<CancellationException> caller = new CancelSource<>(CancellationException::new);
+    AtomicReference<CancellationToken<CancellationException>> observed = new AtomicReference<>();
+    ExternalStoragePayloadTransformer transformer =
+        ExternalStoragePayloadTransformer.fromOptions(
+            ExternalStorageOptions.newBuilder()
+                .setDriver(driver)
+                .setDriverSelector(
+                    (context, payload) -> {
+                      observed.set(context.getCancellationToken());
+                      return driver;
+                    })
+                .setPayloadSizeThreshold(0)
+                .build());
+
+    transformer.store(Collections.singletonList(payload("a")), null, caller.token());
+
+    assertSame(caller.token(), observed.get());
+  }
+
+  private static ExternalStoragePayloadTransformer transformer(
+      StorageDriver driver, int threshold) {
+    return ExternalStoragePayloadTransformer.fromOptions(
+        ExternalStorageOptions.newBuilder()
+            .setDriver(driver)
+            .setPayloadSizeThreshold(threshold)
+            .build());
+  }
+
+  private static Payload payload(String data) {
+    return Payload.newBuilder().setData(ByteString.copyFromUtf8(data)).build();
+  }
+
+  private static String repeat(String s, int n) {
+    StringBuilder sb = new StringBuilder(s.length() * n);
+    for (int i = 0; i < n; i++) {
+      sb.append(s);
+    }
+    return sb.toString();
+  }
+
+  private static Throwable causeOf(CompletableFuture<?> future) {
+    try {
+      future.get();
+      fail("expected failure");
+      return null;
+    } catch (ExecutionException e) {
+      return e.getCause();
+    } catch (InterruptedException e) {
+      Thread.currentThread().interrupt();
+      throw new RuntimeException(e);
+    }
+  }
+
+  private static StorageDriver controlledStore(
+      String name, CompletableFuture<List<StorageDriverClaim>> future) {
+    return new FakeDriver(name) {
+      @Override
+      public CompletableFuture<List<StorageDriverClaim>> store(
+          StorageDriverStoreContext context, List<Payload> payloads) {
+        return future;
+      }
+    };
+  }
+
+  private static class FakeDriver implements StorageDriver {
+    private final String name;
+
+    FakeDriver(String name) {
+      this.name = name;
+    }
+
+    @Override
+    public String getName() {
+      return name;
+    }
+
+    @Override
+    public String getType() {
+      return "test.fake";
+    }
+
+    @Override
+    public CompletableFuture<List<StorageDriverClaim>> store(
+        StorageDriverStoreContext context, List<Payload> payloads) {
+      throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public CompletableFuture<List<Payload>> retrieve(
+        StorageDriverRetrieveContext context, List<StorageDriverClaim> claims) {
+      throw new UnsupportedOperationException();
+    }
+  }
+
+  private static class InMemoryDriver extends FakeDriver {
+    final Map<String, Payload> objects = new HashMap<>();
+    final List<Integer> storeBatchSizes = new ArrayList<>();
+    final List<Integer> retrieveBatchSizes = new ArrayList<>();
+    private int counter = 0;
+
+    InMemoryDriver(String name) {
+      super(name);
+    }
+
+    @Override
+    public CompletableFuture<List<StorageDriverClaim>> store(
+        StorageDriverStoreContext context, List<Payload> payloads) {
+      storeBatchSizes.add(payloads.size());
+      List<StorageDriverClaim> claims = new ArrayList<>();
+      for (Payload payload : payloads) {
+        String key = getName() + "-" + (counter++);
+        objects.put(key, payload);
+        claims.add(new StorageDriverClaim(Collections.singletonMap("key", key)));
+      }
+      return CompletableFuture.completedFuture(claims);
+    }
+
+    @Override
+    public CompletableFuture<List<Payload>> retrieve(
+        StorageDriverRetrieveContext context, List<StorageDriverClaim> claims) {
+      retrieveBatchSizes.add(claims.size());
+      List<Payload> payloads = new ArrayList<>();
+      for (StorageDriverClaim claim : claims) {
+        payloads.add(objects.get(claim.getClaimData().get("key")));
+      }
+      return CompletableFuture.completedFuture(payloads);
+    }
+  }
+}
diff --git a/temporal-sdk/src/test/java/io/temporal/internal/payload/storage/ExternalStorageReferencesTest.java b/temporal-sdk/src/test/java/io/temporal/internal/payload/storage/ExternalStorageReferencesTest.java
new file mode 100644
index 000000000..3f7b6b948
--- /dev/null
+++ b/temporal-sdk/src/test/java/io/temporal/internal/payload/storage/ExternalStorageReferencesTest.java
@@ -0,0 +1,134 @@
+package io.temporal.internal.payload.storage;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+
+import com.google.protobuf.ByteString;
+import io.temporal.api.common.v1.Payload;
+import io.temporal.api.sdk.v1.ExternalStorageReference;
+import io.temporal.common.converter.EncodingKeys;
+import io.temporal.payload.storage.StorageDriverClaim;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import org.junit.Test;
+
+/** Tests external storage reference encoding and decoding. */
+public class ExternalStorageReferencesTest {
+
+  @Test
+  public void currentFormatRoundTrips() {
+    Map<String, String> claimData = new HashMap<>();
+    claimData.put("bucket", "my-bucket");
+    claimData.put("key", "abc123");
+    StorageDriverClaim claim = new StorageDriverClaim(claimData);
+
+    Payload reference = ExternalStorageReferences.toReferencePayload("driver-1", claim, 4096L);
+
+    assertEquals(1, reference.getExternalPayloadsCount());
+    assertEquals(4096L, reference.getExternalPayloads(0).getSizeBytes());
+    assertEquals(
+        "json/protobuf",
+        reference.getMetadataMap().get(EncodingKeys.METADATA_ENCODING_KEY).toStringUtf8());
+
+    ExternalStorageReferences.ParsedReference parsed =
+        ExternalStorageReferences.tryParseReference(reference);
+    assertNotNull(parsed);
+    assertEquals("driver-1", parsed.driverName);
+    assertEquals(claim, parsed.claim);
+  }
+
+  @Test
+  public void inlinePayloadIsNotAReference() {
+    Payload inline =
+        Payload.newBuilder()
+            .putMetadata(EncodingKeys.METADATA_ENCODING_KEY, ByteString.copyFromUtf8("json/plain"))
+            .setData(ByteString.copyFromUtf8("\"hello\""))
+            .build();
+    assertNull(ExternalStorageReferences.tryParseReference(inline));
+  }
+
+  /**
+   * {@code external_payloads} records the original size for the server and is not part of the
+   * exchange contract, so a producer that omits it still writes a readable reference.
+   */
+  @Test
+  public void referenceWithoutExternalPayloadsIsStillAReference() {
+    Payload reference =
+        Payload.newBuilder()
+            .putMetadata(
+                EncodingKeys.METADATA_ENCODING_KEY, ByteString.copyFromUtf8("json/protobuf"))
+            .putMetadata(
+                EncodingKeys.METADATA_MESSAGE_TYPE_KEY,
+                ByteString.copyFromUtf8(ExternalStorageReference.getDescriptor().getFullName()))
+            .setData(ByteString.copyFromUtf8("{\"driverName\":\"driver-1\"}"))
+            .build();
+
+    ExternalStorageReferences.ParsedReference parsed =
+        ExternalStorageReferences.tryParseReference(reference);
+    assertNotNull(parsed);
+    assertEquals("driver-1", parsed.driverName);
+  }
+
+  @Test
+  public void payloadWithReferenceMessageTypeButForeignEncodingIsNotAReference() {
+    Payload foreign =
+        Payload.newBuilder()
+            .putMetadata(EncodingKeys.METADATA_ENCODING_KEY, ByteString.copyFromUtf8("json/plain"))
+            .putMetadata(
+                EncodingKeys.METADATA_MESSAGE_TYPE_KEY,
+                ByteString.copyFromUtf8(ExternalStorageReference.getDescriptor().getFullName()))
+            .setData(ByteString.copyFromUtf8("{\"driverName\":\"driver-1\"}"))
+            .addExternalPayloads(
+                Payload.ExternalPayloadDetails.newBuilder().setSizeBytes(4096L).build())
+            .build();
+
+    assertNull(ExternalStorageReferences.tryParseReference(foreign));
+  }
+
+  @Test
+  public void payloadWithExternalPayloadsButForeignMessageTypeIsNotAReference() {
+    Payload foreign =
+        Payload.newBuilder()
+            .putMetadata(
+                EncodingKeys.METADATA_ENCODING_KEY, ByteString.copyFromUtf8("json/protobuf"))
+            .putMetadata(
+                EncodingKeys.METADATA_MESSAGE_TYPE_KEY,
+                ByteString.copyFromUtf8("some.other.sdk.v1.ExternalStorageReference"))
+            .setData(ByteString.copyFromUtf8("{\"foo\":1}"))
+            .addExternalPayloads(
+                Payload.ExternalPayloadDetails.newBuilder().setSizeBytes(4096L).build())
+            .build();
+
+    assertNull(ExternalStorageReferences.tryParseReference(foreign));
+  }
+
+  /**
+   * References written by other SDKs must stay readable, so parsing tolerates snake_case field
+   * names and fields added to the proto after this release.
+   */
+  @Test
+  public void parsesReferenceWrittenByAnotherSdk() {
+    Payload reference =
+        Payload.newBuilder()
+            .putMetadata(
+                EncodingKeys.METADATA_ENCODING_KEY, ByteString.copyFromUtf8("json/protobuf"))
+            .putMetadata(
+                EncodingKeys.METADATA_MESSAGE_TYPE_KEY,
+                ByteString.copyFromUtf8(ExternalStorageReference.getDescriptor().getFullName()))
+            .setData(
+                ByteString.copyFromUtf8(
+                    "{\"driver_name\":\"driver-1\",\"claim_data\":{\"key\":\"abc123\"},"
+                        + "\"field_added_later\":\"ignored\"}"))
+            .addExternalPayloads(
+                Payload.ExternalPayloadDetails.newBuilder().setSizeBytes(4096L).build())
+            .build();
+
+    ExternalStorageReferences.ParsedReference parsed =
+        ExternalStorageReferences.tryParseReference(reference);
+    assertNotNull(parsed);
+    assertEquals("driver-1", parsed.driverName);
+    assertEquals(new StorageDriverClaim(Collections.singletonMap("key", "abc123")), parsed.claim);
+  }
+}
diff --git a/temporal-sdk/src/test/java/io/temporal/payload/storage/ExternalStorageOptionsTest.java b/temporal-sdk/src/test/java/io/temporal/payload/storage/ExternalStorageOptionsTest.java
index bc2ed1bc5..2c7ffc782 100644
--- a/temporal-sdk/src/test/java/io/temporal/payload/storage/ExternalStorageOptionsTest.java
+++ b/temporal-sdk/src/test/java/io/temporal/payload/storage/ExternalStorageOptionsTest.java
@@ -11,6 +11,7 @@
 import java.util.concurrent.CompletableFuture;
 import org.junit.Test;
 
+/** Tests external storage option validation and defaults. */
 public class ExternalStorageOptionsTest {
 
   private static StorageDriverStoreContext storeContext(StorageDriverTargetInfo target) {