diff --git a/temporalio/lib/temporalio/converters.rb b/temporalio/lib/temporalio/converters.rb
index ba21be0b..42c5aa34 100644
--- a/temporalio/lib/temporalio/converters.rb
+++ b/temporalio/lib/temporalio/converters.rb
@@ -1,6 +1,15 @@
 # frozen_string_literal: true
 
 require 'temporalio/converters/data_converter'
+require 'temporalio/converters/external_storage'
+require 'temporalio/converters/storage_driver'
+require 'temporalio/converters/storage_driver_activity_info'
+require 'temporalio/converters/storage_driver_claim'
+require 'temporalio/converters/storage_driver_retrieve_context'
+require 'temporalio/converters/storage_driver_select_context'
+require 'temporalio/converters/storage_driver_store_context'
+require 'temporalio/converters/storage_driver_workflow_info'
+require 'temporalio/internal/external_storage_references'
 
 module Temporalio
   # Module for data conversion.
diff --git a/temporalio/lib/temporalio/converters/external_storage.rb b/temporalio/lib/temporalio/converters/external_storage.rb
new file mode 100644
index 00000000..861b223e
--- /dev/null
+++ b/temporalio/lib/temporalio/converters/external_storage.rb
@@ -0,0 +1,79 @@
+# frozen_string_literal: true
+
+require 'temporalio/converters/storage_driver'
+
+module Temporalio
+  module Converters
+    # Configuration for offloading large payloads to an external storage system.
+    #
+    # This is validated when constructed rather than when used, so a misconfiguration surfaces at startup instead of on
+    # the first payload large enough to offload.
+    #
+    # @note WARNING: This API is experimental and may change in the future.
+    #
+    # @!visibility private
+    class ExternalStorage
+      # Shared with every other SDK, so the same payload offloads regardless of which one wrote it.
+      DEFAULT_PAYLOAD_SIZE_THRESHOLD = 256 * 1024
+
+      # @return [Array<StorageDriver>] Drivers available for storing and retrieving payloads.
+      attr_reader :drivers
+
+      # @return [Proc] Selector choosing which driver stores each payload. Called with a
+      #   {StorageDriverSelectContext} and a payload, returning a {StorageDriver} or nil to pass the payload through.
+      #   Only called for payloads that meet {#payload_size_threshold}, and must return one of {#drivers} or nil.
+      attr_reader :driver_selector
+
+      # @return [Integer] Minimum encoded payload size, in bytes, that is offloaded. Payloads at or above this size are
+      #   offloaded; smaller ones are left inline. Zero offloads every payload.
+      attr_reader :payload_size_threshold
+
+      # Create external storage configuration.
+      #
+      # @param drivers [Array<StorageDriver>] Drivers available for storing and retrieving payloads. At least one is
+      #   required, and every driver must have a unique, non-empty name. Retrieval routes by the name recorded in
+      #   history, so a driver must remain configured for as long as any payload it stored can still be read.
+      # @param driver_selector [Proc, nil] Selector choosing which driver stores each payload. Required when more than
+      #   one driver is given; with a single driver it may be nil, in which case that driver stores every eligible
+      #   payload.
+      # @param payload_size_threshold [Integer] Minimum encoded payload size, in bytes, to offload. Must be a
+      #   non-negative Integer.
+      def initialize(drivers:, driver_selector: nil, payload_size_threshold: DEFAULT_PAYLOAD_SIZE_THRESHOLD)
+        raise ArgumentError, 'At least one driver is required' if drivers.empty?
+
+        unless payload_size_threshold.is_a?(Integer) && payload_size_threshold >= 0
+          raise ArgumentError, 'payload_size_threshold must be a non-negative Integer'
+        end
+
+        if drivers.size > 1 && driver_selector.nil?
+          raise ArgumentError, 'driver_selector is required when more than one driver is given'
+        end
+
+        @drivers_by_name = {}
+        drivers.each do |driver|
+          name = driver.name
+          # The name is the routing key written into history and is written to a string proto field, so anything but
+          # a non-empty String cannot be resolved on read. A Symbol in particular passes a bare empty? check.
+          raise ArgumentError, 'Driver name must be a non-empty String' unless name.is_a?(String) && !name.empty?
+          raise ArgumentError, "Multiple drivers given with name '#{name}'" if @drivers_by_name.key?(name)
+
+          @drivers_by_name[name] = driver
+        end
+        @drivers_by_name.freeze
+
+        @drivers = drivers.dup.freeze
+        # Normalized so callers never have to distinguish the single-driver case.
+        @driver_selector = driver_selector || ->(_context, _payload) { @drivers.first }
+        @payload_size_threshold = payload_size_threshold
+      end
+
+      # Get the driver registered under the given name.
+      #
+      # @param name [String] Driver name recorded in the payload reference.
+      # @return [StorageDriver, nil] The driver, or nil if none is registered under that name.
+      def driver(name)
+        @drivers_by_name[name]
+      end
+    end
+  end
+end
diff --git a/temporalio/lib/temporalio/converters/storage_driver.rb b/temporalio/lib/temporalio/converters/storage_driver.rb
new file mode 100644
index 00000000..0193831f
--- /dev/null
+++ b/temporalio/lib/temporalio/converters/storage_driver.rb
@@ -0,0 +1,56 @@
+# frozen_string_literal: true
+
+module Temporalio
+  module Converters
+    # Base class for storing and retrieving payloads in an external storage system.
+    #
+    # Payloads at or over the configured size threshold are handed to a driver and replaced on the wire by a small
+    # reference, so the payload data itself never reaches the Temporal server.
+    #
+    # Implementations are called concurrently and must be thread safe. Methods block, and may perform I/O directly.
+    # To work a batch concurrently, fan out with threads: threads overlap blocking I/O, since Ruby releases its global
+    # lock around it, and they work whether or not the worker is running under a fiber scheduler.
+    # {::Fiber.schedule} must not be used, since it raises when the worker is configured with a payload codec thread
+    # pool instead of a fiber scheduler. {Temporalio::Worker::ThreadPool} can be used to bound that fan out.
+    #
+    # @note WARNING: This API is experimental and may change in the future.
+    #
+    # @!visibility private
+    class StorageDriver
+      # Name of this driver instance, unique among the drivers registered on one {ExternalStorage}. This is written
+      # into history alongside every payload the driver stores and is used to route retrieval back to the same driver.
+      # Renaming a deployed driver makes every payload stored under the old name unretrievable.
+      #
+      # @return [String] Driver instance name.
+      def name
+        raise NotImplementedError
+      end
+
+      # Identifier for this driver implementation, for example +aws.s3driver+. Unlike {#name}, this is identical across
+      # every instance of the same implementation and across SDK languages.
+      #
+      # @return [String] Driver implementation identifier.
+      def type
+        raise NotImplementedError
+      end
+
+      # Store the given payloads.
+      #
+      # @param context [StorageDriverStoreContext] Context for this store operation.
+      # @param payloads [Enumerable<Api::Common::V1::Payload>] Payloads to store. This value should not be mutated.
+      # @return [Array<StorageDriverClaim>] One claim per payload, in the same order.
+      def store(context, payloads)
+        raise NotImplementedError
+      end
+
+      # Retrieve the payloads for the given claims.
+      #
+      # @param context [StorageDriverRetrieveContext] Context for this retrieve operation.
+      # @param claims [Enumerable<StorageDriverClaim>] Claims to retrieve. This value should not be mutated.
+      # @return [Array<Api::Common::V1::Payload>] One payload per claim, in the same order.
+      def retrieve(context, claims)
+        raise NotImplementedError
+      end
+    end
+  end
+end
diff --git a/temporalio/lib/temporalio/converters/storage_driver_activity_info.rb b/temporalio/lib/temporalio/converters/storage_driver_activity_info.rb
new file mode 100644
index 00000000..b16456cf
--- /dev/null
+++ b/temporalio/lib/temporalio/converters/storage_driver_activity_info.rb
@@ -0,0 +1,38 @@
+# frozen_string_literal: true
+
+module Temporalio
+  module Converters
+    StorageDriverActivityInfo = Data.define(
+      :namespace,
+      :id,
+      :run_id,
+      :type
+    )
+
+    # Identity of a standalone activity a payload is being stored on behalf of.
+    #
+    # @note WARNING: This API is experimental and may change in the future.
+    #
+    # @!attribute [r] namespace
+    #   @return [String] Activity namespace.
+    # @!attribute [r] id
+    #   @return [String, nil] Activity ID, if known.
+    # @!attribute [r] run_id
+    #   @return [String, nil] Run ID, if known.
+    # @!attribute [r] type
+    #   @return [String, nil] Activity type name, if known.
+    #
+    # @!visibility private
+    class StorageDriverActivityInfo
+      # Create activity target information.
+      #
+      # @param namespace [String] Activity namespace.
+      # @param id [String, nil] Activity ID.
+      # @param run_id [String, nil] Run ID.
+      # @param type [String, nil] Activity type name.
+      def initialize(namespace:, id: nil, run_id: nil, type: nil)
+        super
+      end
+    end
+  end
+end
diff --git a/temporalio/lib/temporalio/converters/storage_driver_claim.rb b/temporalio/lib/temporalio/converters/storage_driver_claim.rb
new file mode 100644
index 00000000..bdc85bad
--- /dev/null
+++ b/temporalio/lib/temporalio/converters/storage_driver_claim.rb
@@ -0,0 +1,30 @@
+# frozen_string_literal: true
+
+module Temporalio
+  module Converters
+    StorageDriverClaim = Data.define(:claim_data)
+
+    # Driver-defined reference to an externally stored payload, used to retrieve it later.
+    #
+    # @note WARNING: This API is experimental and may change in the future.
+    #
+    # @!attribute [r] claim_data
+    #   @return [Hash<String, String>] Data the driver needs to retrieve the payload. This is written into history, so
+    #     it must contain everything required for retrieval and must not contain secrets.
+    #
+    # @!visibility private
+    class StorageDriverClaim
+      # Create a claim.
+      #
+      # @param claim_data [Hash<String, String>] Data identifying the stored payload.
+      def initialize(claim_data:)
+        # Copied and frozen because a claim is compared and hashed by this data, so a later mutation would silently
+        # change its identity. The values are copied individually because freezing the Hash leaves them mutable in
+        # place and still aliased to the caller's objects.
+        # steep:ignore:start
+        super(claim_data: claim_data.transform_values { |value| value.dup.freeze }.freeze)
+        # steep:ignore:end
+      end
+    end
+  end
+end
diff --git a/temporalio/lib/temporalio/converters/storage_driver_retrieve_context.rb b/temporalio/lib/temporalio/converters/storage_driver_retrieve_context.rb
new file mode 100644
index 00000000..412384eb
--- /dev/null
+++ b/temporalio/lib/temporalio/converters/storage_driver_retrieve_context.rb
@@ -0,0 +1,23 @@
+# frozen_string_literal: true
+
+require 'temporalio/cancellation'
+
+module Temporalio
+  module Converters
+    StorageDriverRetrieveContext = Data.define(:cancellation)
+
+    # Context given to {StorageDriver#retrieve}.
+    #
+    # This deliberately carries no execution target. A payload must be retrievable from its {StorageDriverClaim}
+    # alone, since the same payload can be read from a different execution than the one that stored it.
+    #
+    # @note WARNING: This API is experimental and may change in the future. Members may be added, so accept keyword
+    #   arguments defensively if constructing this yourself.
+    #
+    # @!attribute [r] cancellation
+    #   @return [Cancellation] Cancelled when the SDK abandons this retrieve operation.
+    #
+    # @!visibility private
+    class StorageDriverRetrieveContext; end # rubocop:disable Lint/EmptyClass
+  end
+end
diff --git a/temporalio/lib/temporalio/converters/storage_driver_select_context.rb b/temporalio/lib/temporalio/converters/storage_driver_select_context.rb
new file mode 100644
index 00000000..a5ef5be7
--- /dev/null
+++ b/temporalio/lib/temporalio/converters/storage_driver_select_context.rb
@@ -0,0 +1,26 @@
+# frozen_string_literal: true
+
+require 'temporalio/cancellation'
+
+module Temporalio
+  module Converters
+    StorageDriverSelectContext = Data.define(
+      :target,
+      :cancellation
+    )
+
+    # Context given to the driver selector.
+    #
+    # @note WARNING: This API is experimental and may change in the future. Members may be added, so accept keyword
+    #   arguments defensively if constructing this yourself.
+    #
+    # @!attribute [r] target
+    #   @return [StorageDriverWorkflowInfo, StorageDriverActivityInfo, nil] The execution the payload is being stored
+    #     on behalf of, or nil when there is no associated execution.
+    # @!attribute [r] cancellation
+    #   @return [Cancellation] Cancelled when the SDK abandons the operation this selection is part of.
+    #
+    # @!visibility private
+    class StorageDriverSelectContext; end # rubocop:disable Lint/EmptyClass
+  end
+end
diff --git a/temporalio/lib/temporalio/converters/storage_driver_store_context.rb b/temporalio/lib/temporalio/converters/storage_driver_store_context.rb
new file mode 100644
index 00000000..e50f3f76
--- /dev/null
+++ b/temporalio/lib/temporalio/converters/storage_driver_store_context.rb
@@ -0,0 +1,26 @@
+# frozen_string_literal: true
+
+require 'temporalio/cancellation'
+
+module Temporalio
+  module Converters
+    StorageDriverStoreContext = Data.define(
+      :target,
+      :cancellation
+    )
+
+    # Context given to {StorageDriver#store}.
+    #
+    # @note WARNING: This API is experimental and may change in the future. Members may be added, so accept keyword
+    #   arguments defensively if constructing this yourself.
+    #
+    # @!attribute [r] target
+    #   @return [StorageDriverWorkflowInfo, StorageDriverActivityInfo, nil] The execution the payloads are being stored
+    #     on behalf of, or nil when there is no associated execution.
+    # @!attribute [r] cancellation
+    #   @return [Cancellation] Cancelled when the SDK abandons this store operation.
+    #
+    # @!visibility private
+    class StorageDriverStoreContext; end # rubocop:disable Lint/EmptyClass
+  end
+end
diff --git a/temporalio/lib/temporalio/converters/storage_driver_workflow_info.rb b/temporalio/lib/temporalio/converters/storage_driver_workflow_info.rb
new file mode 100644
index 00000000..6a8c8381
--- /dev/null
+++ b/temporalio/lib/temporalio/converters/storage_driver_workflow_info.rb
@@ -0,0 +1,40 @@
+# frozen_string_literal: true
+
+module Temporalio
+  module Converters
+    StorageDriverWorkflowInfo = Data.define(
+      :namespace,
+      :id,
+      :run_id,
+      :type
+    )
+
+    # Identity of the workflow a payload is being stored on behalf of. Also used for workflow activities, which
+    # store against their owning workflow rather than themselves.
+    #
+    # @note WARNING: This API is experimental and may change in the future.
+    #
+    # @!attribute [r] namespace
+    #   @return [String] Workflow namespace.
+    # @!attribute [r] id
+    #   @return [String, nil] Workflow ID, if known.
+    # @!attribute [r] run_id
+    #   @return [String, nil] Run ID, if known. Unset when the run is not yet determined, such as when starting a child
+    #     workflow or continuing as new.
+    # @!attribute [r] type
+    #   @return [String, nil] Workflow type name, if known.
+    #
+    # @!visibility private
+    class StorageDriverWorkflowInfo
+      # Create workflow target information.
+      #
+      # @param namespace [String] Workflow namespace.
+      # @param id [String, nil] Workflow ID.
+      # @param run_id [String, nil] Run ID.
+      # @param type [String, nil] Workflow type name.
+      def initialize(namespace:, id: nil, run_id: nil, type: nil)
+        super
+      end
+    end
+  end
+end
diff --git a/temporalio/lib/temporalio/error.rb b/temporalio/lib/temporalio/error.rb
index 8d967864..e1d4c171 100644
--- a/temporalio/lib/temporalio/error.rb
+++ b/temporalio/lib/temporalio/error.rb
@@ -116,6 +116,23 @@ def initialize(details)
       end
     end
 
+    # Error that occurs when a payload stored in external storage is encountered but no external storage is
+    # configured to retrieve it. This usually means the client or worker reading the payload is configured
+    # differently from the one that wrote it.
+    #
+    # @note WARNING: This API is experimental and may change in the future.
+    #
+    # @!visibility private
+    class ExternalStorageNotConfiguredError < Error
+      # @!visibility private
+      def initialize(
+        message = '[TMPRL1105] Encountered a reference to a payload in external storage, but no external storage ' \
+                  'is configured to retrieve it. Configure external storage with a driver able to retrieve it.'
+      )
+        super
+      end
+    end
+
     # Error raised by a client for a general RPC failure.
     class RPCError < Error
       # @return [Code] Status code for the error.
diff --git a/temporalio/lib/temporalio/internal/external_storage_references.rb b/temporalio/lib/temporalio/internal/external_storage_references.rb
new file mode 100644
index 00000000..15502b22
--- /dev/null
+++ b/temporalio/lib/temporalio/internal/external_storage_references.rb
@@ -0,0 +1,62 @@
+# frozen_string_literal: true
+
+require 'temporalio/api'
+require 'temporalio/api/sdk/v1/external_storage'
+
+module Temporalio
+  module Internal
+    # The on-the-wire form of a payload that has been offloaded to external storage.
+    #
+    # An offloaded payload is replaced by one whose data is the proto-JSON form of an
+    # +ExternalStorageReference+. This shape is part of the wire contract and must not be changed unilaterally.
+    #
+    # @!visibility private
+    module ExternalStorageReferences
+      ENCODING = 'json/protobuf'
+      MESSAGE_TYPE = 'temporal.api.sdk.v1.ExternalStorageReference'
+
+      class << self
+        # @param payload [Api::Common::V1::Payload] Payload to check.
+        # @return [Boolean] Whether the payload is a reference to externally stored data.
+        def reference?(payload)
+          payload.metadata['encoding'] == ENCODING && payload.metadata['messageType'] == MESSAGE_TYPE
+        end
+
+        # @param payload [Api::Common::V1::Payload] Payload to parse.
+        # @return [Api::Sdk::V1::ExternalStorageReference, nil] Parsed reference, or nil if not a reference.
+        #
+        # TODO: Surface parse failures as a Temporal error instead of Google::Protobuf::ParseError. The runner is the
+        # right layer for it, since only the caller knows which payload failed; a malformed reference cannot name its
+        # own driver.
+        def parse_reference(payload)
+          return nil unless reference?(payload)
+
+          # Another SDK may add fields to the reference before this one knows about them, and an unknown field must
+          # not make an otherwise-valid payload unreadable.
+          Api::Sdk::V1::ExternalStorageReference.decode_json(
+            payload.data, ignore_unknown_fields: true
+          )
+        end
+
+        # @param driver_name [String] Name of the driver that stored the payload.
+        # @param claim_data [Hash<String, String>] Driver data identifying the stored payload.
+        # @param original_size_bytes [Integer] Encoded size of the payload that was offloaded.
+        # @return [Api::Common::V1::Payload] Reference payload that replaces the offloaded payload on the wire.
+        def create_reference_payload(driver_name:, claim_data:, original_size_bytes:)
+          reference = Api::Sdk::V1::ExternalStorageReference.new(
+            driver_name:, claim_data: claim_data.to_h
+          )
+          Api::Common::V1::Payload.new(
+            metadata: { 'encoding' => ENCODING.b, 'messageType' => MESSAGE_TYPE.b },
+            data: reference.to_json.b,
+            # The original size is kept so the server and UI can report what the payload would have been without
+            # having to fetch it from storage.
+            external_payloads: [
+              Api::Common::V1::Payload::ExternalPayloadDetails.new(size_bytes: original_size_bytes)
+            ]
+          )
+        end
+      end
+    end
+  end
+end
diff --git a/temporalio/rbi/temporalio/converters/external_storage.rbi b/temporalio/rbi/temporalio/converters/external_storage.rbi
new file mode 100644
index 00000000..818b314e
--- /dev/null
+++ b/temporalio/rbi/temporalio/converters/external_storage.rbi
@@ -0,0 +1,40 @@
+# typed: true
+
+class Temporalio::Converters::ExternalStorage
+  extend T::Sig
+
+  DEFAULT_PAYLOAD_SIZE_THRESHOLD = T.let(T.unsafe(nil), Integer)
+
+  sig { returns(T::Array[Temporalio::Converters::StorageDriver]) }
+  attr_reader :drivers
+
+  sig do
+    returns(
+      T.proc.params(
+        context: Temporalio::Converters::StorageDriverSelectContext,
+        payload: Temporalio::Api::Common::V1::Payload
+      ).returns(T.nilable(Temporalio::Converters::StorageDriver))
+    )
+  end
+  attr_reader :driver_selector
+
+  sig { returns(Integer) }
+  attr_reader :payload_size_threshold
+
+  sig do
+    params(
+      drivers: T::Array[Temporalio::Converters::StorageDriver],
+      driver_selector: T.nilable(
+        T.proc.params(
+          context: Temporalio::Converters::StorageDriverSelectContext,
+          payload: Temporalio::Api::Common::V1::Payload
+        ).returns(T.nilable(Temporalio::Converters::StorageDriver))
+      ),
+      payload_size_threshold: Integer
+    ).void
+  end
+  def initialize(drivers:, driver_selector: T.unsafe(nil), payload_size_threshold: T.unsafe(nil)); end
+
+  sig { params(name: String).returns(T.nilable(Temporalio::Converters::StorageDriver)) }
+  def driver(name); end
+end
diff --git a/temporalio/rbi/temporalio/converters/storage_driver.rbi b/temporalio/rbi/temporalio/converters/storage_driver.rbi
new file mode 100644
index 00000000..62f645ee
--- /dev/null
+++ b/temporalio/rbi/temporalio/converters/storage_driver.rbi
@@ -0,0 +1,27 @@
+# typed: true
+
+class Temporalio::Converters::StorageDriver
+  extend T::Sig
+
+  sig { returns(String) }
+  def name; end
+
+  sig { returns(String) }
+  def type; end
+
+  sig do
+    params(
+      context: Temporalio::Converters::StorageDriverStoreContext,
+      payloads: T::Enumerable[Temporalio::Api::Common::V1::Payload]
+    ).returns(T::Array[Temporalio::Converters::StorageDriverClaim])
+  end
+  def store(context, payloads); end
+
+  sig do
+    params(
+      context: Temporalio::Converters::StorageDriverRetrieveContext,
+      claims: T::Enumerable[Temporalio::Converters::StorageDriverClaim]
+    ).returns(T::Array[Temporalio::Api::Common::V1::Payload])
+  end
+  def retrieve(context, claims); end
+end
diff --git a/temporalio/rbi/temporalio/converters/storage_driver_activity_info.rbi b/temporalio/rbi/temporalio/converters/storage_driver_activity_info.rbi
new file mode 100644
index 00000000..89b1ca48
--- /dev/null
+++ b/temporalio/rbi/temporalio/converters/storage_driver_activity_info.rbi
@@ -0,0 +1,36 @@
+# typed: true
+
+class Temporalio::Converters::StorageDriverActivityInfo < ::Data
+  sig do
+    params(
+      namespace: String,
+      id: T.nilable(String),
+      run_id: T.nilable(String),
+      type: T.nilable(String)
+    ).void
+  end
+  def initialize(namespace:, id: T.unsafe(nil), run_id: T.unsafe(nil), type: T.unsafe(nil)); end
+
+  sig { returns(String) }
+  def namespace; end
+
+  sig { returns(T.nilable(String)) }
+  def id; end
+
+  sig { returns(T.nilable(String)) }
+  def run_id; end
+
+  sig { returns(T.nilable(String)) }
+  def type; end
+
+  class << self
+    sig { params(args: T.untyped).returns(Temporalio::Converters::StorageDriverActivityInfo) }
+    def [](*args); end
+
+    sig { returns(T::Array[Symbol]) }
+    def members; end
+
+    sig { params(args: T.untyped).returns(Temporalio::Converters::StorageDriverActivityInfo) }
+    def new(*args); end
+  end
+end
diff --git a/temporalio/rbi/temporalio/converters/storage_driver_claim.rbi b/temporalio/rbi/temporalio/converters/storage_driver_claim.rbi
new file mode 100644
index 00000000..1aefbd97
--- /dev/null
+++ b/temporalio/rbi/temporalio/converters/storage_driver_claim.rbi
@@ -0,0 +1,20 @@
+# typed: true
+
+class Temporalio::Converters::StorageDriverClaim < ::Data
+  sig { params(claim_data: T::Hash[String, String]).void }
+  def initialize(claim_data:); end
+
+  sig { returns(T::Hash[String, String]) }
+  def claim_data; end
+
+  class << self
+    sig { params(args: T.untyped).returns(Temporalio::Converters::StorageDriverClaim) }
+    def [](*args); end
+
+    sig { returns(T::Array[Symbol]) }
+    def members; end
+
+    sig { params(args: T.untyped).returns(Temporalio::Converters::StorageDriverClaim) }
+    def new(*args); end
+  end
+end
diff --git a/temporalio/rbi/temporalio/converters/storage_driver_retrieve_context.rbi b/temporalio/rbi/temporalio/converters/storage_driver_retrieve_context.rbi
new file mode 100644
index 00000000..0dee3622
--- /dev/null
+++ b/temporalio/rbi/temporalio/converters/storage_driver_retrieve_context.rbi
@@ -0,0 +1,20 @@
+# typed: true
+
+class Temporalio::Converters::StorageDriverRetrieveContext < ::Data
+  sig { params(cancellation: Temporalio::Cancellation).void }
+  def initialize(cancellation:); end
+
+  sig { returns(Temporalio::Cancellation) }
+  def cancellation; end
+
+  class << self
+    sig { params(args: T.untyped).returns(Temporalio::Converters::StorageDriverRetrieveContext) }
+    def [](*args); end
+
+    sig { returns(T::Array[Symbol]) }
+    def members; end
+
+    sig { params(args: T.untyped).returns(Temporalio::Converters::StorageDriverRetrieveContext) }
+    def new(*args); end
+  end
+end
diff --git a/temporalio/rbi/temporalio/converters/storage_driver_select_context.rbi b/temporalio/rbi/temporalio/converters/storage_driver_select_context.rbi
new file mode 100644
index 00000000..b6b9ef9f
--- /dev/null
+++ b/temporalio/rbi/temporalio/converters/storage_driver_select_context.rbi
@@ -0,0 +1,42 @@
+# typed: true
+
+class Temporalio::Converters::StorageDriverSelectContext < ::Data
+  sig do
+    params(
+      target: T.nilable(
+        T.any(
+          Temporalio::Converters::StorageDriverWorkflowInfo,
+          Temporalio::Converters::StorageDriverActivityInfo
+        )
+      ),
+      cancellation: Temporalio::Cancellation
+    ).void
+  end
+  def initialize(target:, cancellation:); end
+
+  sig do
+    returns(
+      T.nilable(
+        T.any(
+          Temporalio::Converters::StorageDriverWorkflowInfo,
+          Temporalio::Converters::StorageDriverActivityInfo
+        )
+      )
+    )
+  end
+  def target; end
+
+  sig { returns(Temporalio::Cancellation) }
+  def cancellation; end
+
+  class << self
+    sig { params(args: T.untyped).returns(Temporalio::Converters::StorageDriverSelectContext) }
+    def [](*args); end
+
+    sig { returns(T::Array[Symbol]) }
+    def members; end
+
+    sig { params(args: T.untyped).returns(Temporalio::Converters::StorageDriverSelectContext) }
+    def new(*args); end
+  end
+end
diff --git a/temporalio/rbi/temporalio/converters/storage_driver_store_context.rbi b/temporalio/rbi/temporalio/converters/storage_driver_store_context.rbi
new file mode 100644
index 00000000..b4da1088
--- /dev/null
+++ b/temporalio/rbi/temporalio/converters/storage_driver_store_context.rbi
@@ -0,0 +1,42 @@
+# typed: true
+
+class Temporalio::Converters::StorageDriverStoreContext < ::Data
+  sig do
+    params(
+      target: T.nilable(
+        T.any(
+          Temporalio::Converters::StorageDriverWorkflowInfo,
+          Temporalio::Converters::StorageDriverActivityInfo
+        )
+      ),
+      cancellation: Temporalio::Cancellation
+    ).void
+  end
+  def initialize(target:, cancellation:); end
+
+  sig do
+    returns(
+      T.nilable(
+        T.any(
+          Temporalio::Converters::StorageDriverWorkflowInfo,
+          Temporalio::Converters::StorageDriverActivityInfo
+        )
+      )
+    )
+  end
+  def target; end
+
+  sig { returns(Temporalio::Cancellation) }
+  def cancellation; end
+
+  class << self
+    sig { params(args: T.untyped).returns(Temporalio::Converters::StorageDriverStoreContext) }
+    def [](*args); end
+
+    sig { returns(T::Array[Symbol]) }
+    def members; end
+
+    sig { params(args: T.untyped).returns(Temporalio::Converters::StorageDriverStoreContext) }
+    def new(*args); end
+  end
+end
diff --git a/temporalio/rbi/temporalio/converters/storage_driver_workflow_info.rbi b/temporalio/rbi/temporalio/converters/storage_driver_workflow_info.rbi
new file mode 100644
index 00000000..e8455172
--- /dev/null
+++ b/temporalio/rbi/temporalio/converters/storage_driver_workflow_info.rbi
@@ -0,0 +1,36 @@
+# typed: true
+
+class Temporalio::Converters::StorageDriverWorkflowInfo < ::Data
+  sig do
+    params(
+      namespace: String,
+      id: T.nilable(String),
+      run_id: T.nilable(String),
+      type: T.nilable(String)
+    ).void
+  end
+  def initialize(namespace:, id: T.unsafe(nil), run_id: T.unsafe(nil), type: T.unsafe(nil)); end
+
+  sig { returns(String) }
+  def namespace; end
+
+  sig { returns(T.nilable(String)) }
+  def id; end
+
+  sig { returns(T.nilable(String)) }
+  def run_id; end
+
+  sig { returns(T.nilable(String)) }
+  def type; end
+
+  class << self
+    sig { params(args: T.untyped).returns(Temporalio::Converters::StorageDriverWorkflowInfo) }
+    def [](*args); end
+
+    sig { returns(T::Array[Symbol]) }
+    def members; end
+
+    sig { params(args: T.untyped).returns(Temporalio::Converters::StorageDriverWorkflowInfo) }
+    def new(*args); end
+  end
+end
diff --git a/temporalio/rbi/temporalio/error.rbi b/temporalio/rbi/temporalio/error.rbi
index ad2b41ad..cec26a2b 100644
--- a/temporalio/rbi/temporalio/error.rbi
+++ b/temporalio/rbi/temporalio/error.rbi
@@ -281,6 +281,13 @@ class Temporalio::Error::ScheduleAlreadyRunningError < ::Temporalio::Error
   def initialize; end
 end
 
+class Temporalio::Error::ExternalStorageNotConfiguredError < ::Temporalio::Error
+  extend T::Sig
+
+  sig { params(message: String).void }
+  def initialize(message = T.unsafe(nil)); end
+end
+
 class Temporalio::Error::RPCError < ::Temporalio::Error
   sig { params(message: String, code: Integer, raw_grpc_status: T.nilable(Object)).void }
   def initialize(message, code:, raw_grpc_status:); end
diff --git a/temporalio/rbi/temporalio/internal/external_storage_references.rbi b/temporalio/rbi/temporalio/internal/external_storage_references.rbi
new file mode 100644
index 00000000..dc9905dc
--- /dev/null
+++ b/temporalio/rbi/temporalio/internal/external_storage_references.rbi
@@ -0,0 +1,26 @@
+# typed: true
+
+module Temporalio::Internal::ExternalStorageReferences
+  extend T::Sig
+
+  ENCODING = T.let(T.unsafe(nil), String)
+  MESSAGE_TYPE = T.let(T.unsafe(nil), String)
+
+  sig { params(payload: Temporalio::Api::Common::V1::Payload).returns(T::Boolean) }
+  def self.reference?(payload); end
+
+  sig do
+    params(payload: Temporalio::Api::Common::V1::Payload)
+      .returns(T.nilable(Temporalio::Api::Sdk::V1::ExternalStorageReference))
+  end
+  def self.parse_reference(payload); end
+
+  sig do
+    params(
+      driver_name: String,
+      claim_data: T::Hash[String, String],
+      original_size_bytes: Integer
+    ).returns(Temporalio::Api::Common::V1::Payload)
+  end
+  def self.create_reference_payload(driver_name:, claim_data:, original_size_bytes:); end
+end
diff --git a/temporalio/sig/temporalio/converters/external_storage.rbs b/temporalio/sig/temporalio/converters/external_storage.rbs
new file mode 100644
index 00000000..c903d59a
--- /dev/null
+++ b/temporalio/sig/temporalio/converters/external_storage.rbs
@@ -0,0 +1,23 @@
+module Temporalio
+  module Converters
+    class ExternalStorage
+      DEFAULT_PAYLOAD_SIZE_THRESHOLD: Integer
+
+      type driverSelector = ^(StorageDriverSelectContext, Api::Common::V1::Payload) -> StorageDriver?
+
+      attr_reader drivers: Array[StorageDriver]
+      attr_reader driver_selector: driverSelector
+      attr_reader payload_size_threshold: Integer
+
+      @drivers_by_name: Hash[String, StorageDriver]
+
+      def initialize: (
+        drivers: Array[StorageDriver],
+        ?driver_selector: driverSelector?,
+        ?payload_size_threshold: Integer
+      ) -> void
+
+      def driver: (String name) -> StorageDriver?
+    end
+  end
+end
diff --git a/temporalio/sig/temporalio/converters/storage_driver.rbs b/temporalio/sig/temporalio/converters/storage_driver.rbs
new file mode 100644
index 00000000..79c5594e
--- /dev/null
+++ b/temporalio/sig/temporalio/converters/storage_driver.rbs
@@ -0,0 +1,16 @@
+module Temporalio
+  module Converters
+    class StorageDriver
+      def name: -> String
+      def type: -> String
+      def store: (
+        StorageDriverStoreContext context,
+        Enumerable[Api::Common::V1::Payload] payloads
+      ) -> Array[StorageDriverClaim]
+      def retrieve: (
+        StorageDriverRetrieveContext context,
+        Enumerable[StorageDriverClaim] claims
+      ) -> Array[Api::Common::V1::Payload]
+    end
+  end
+end
diff --git a/temporalio/sig/temporalio/converters/storage_driver_activity_info.rbs b/temporalio/sig/temporalio/converters/storage_driver_activity_info.rbs
new file mode 100644
index 00000000..4ddb563d
--- /dev/null
+++ b/temporalio/sig/temporalio/converters/storage_driver_activity_info.rbs
@@ -0,0 +1,17 @@
+module Temporalio
+  module Converters
+    class StorageDriverActivityInfo
+      attr_reader namespace: String
+      attr_reader id: String?
+      attr_reader run_id: String?
+      attr_reader type: String?
+
+      def initialize: (
+        namespace: String,
+        ?id: String?,
+        ?run_id: String?,
+        ?type: String?
+      ) -> void
+    end
+  end
+end
diff --git a/temporalio/sig/temporalio/converters/storage_driver_claim.rbs b/temporalio/sig/temporalio/converters/storage_driver_claim.rbs
new file mode 100644
index 00000000..5b9deb70
--- /dev/null
+++ b/temporalio/sig/temporalio/converters/storage_driver_claim.rbs
@@ -0,0 +1,9 @@
+module Temporalio
+  module Converters
+    class StorageDriverClaim
+      attr_reader claim_data: Hash[String, String]
+
+      def initialize: (claim_data: Hash[String, String]) -> void
+    end
+  end
+end
diff --git a/temporalio/sig/temporalio/converters/storage_driver_retrieve_context.rbs b/temporalio/sig/temporalio/converters/storage_driver_retrieve_context.rbs
new file mode 100644
index 00000000..90b67f6c
--- /dev/null
+++ b/temporalio/sig/temporalio/converters/storage_driver_retrieve_context.rbs
@@ -0,0 +1,9 @@
+module Temporalio
+  module Converters
+    class StorageDriverRetrieveContext
+      attr_reader cancellation: Cancellation
+
+      def initialize: (cancellation: Cancellation) -> void
+    end
+  end
+end
diff --git a/temporalio/sig/temporalio/converters/storage_driver_select_context.rbs b/temporalio/sig/temporalio/converters/storage_driver_select_context.rbs
new file mode 100644
index 00000000..e4b8aa4c
--- /dev/null
+++ b/temporalio/sig/temporalio/converters/storage_driver_select_context.rbs
@@ -0,0 +1,13 @@
+module Temporalio
+  module Converters
+    class StorageDriverSelectContext
+      attr_reader target: storageDriverTargetInfo?
+      attr_reader cancellation: Cancellation
+
+      def initialize: (
+        target: storageDriverTargetInfo?,
+        cancellation: Cancellation
+      ) -> void
+    end
+  end
+end
diff --git a/temporalio/sig/temporalio/converters/storage_driver_store_context.rbs b/temporalio/sig/temporalio/converters/storage_driver_store_context.rbs
new file mode 100644
index 00000000..a36eceb1
--- /dev/null
+++ b/temporalio/sig/temporalio/converters/storage_driver_store_context.rbs
@@ -0,0 +1,13 @@
+module Temporalio
+  module Converters
+    class StorageDriverStoreContext
+      attr_reader target: storageDriverTargetInfo?
+      attr_reader cancellation: Cancellation
+
+      def initialize: (
+        target: storageDriverTargetInfo?,
+        cancellation: Cancellation
+      ) -> void
+    end
+  end
+end
diff --git a/temporalio/sig/temporalio/converters/storage_driver_target_info.rbs b/temporalio/sig/temporalio/converters/storage_driver_target_info.rbs
new file mode 100644
index 00000000..f0a08633
--- /dev/null
+++ b/temporalio/sig/temporalio/converters/storage_driver_target_info.rbs
@@ -0,0 +1,5 @@
+module Temporalio
+  module Converters
+    type storageDriverTargetInfo = StorageDriverWorkflowInfo | StorageDriverActivityInfo
+  end
+end
diff --git a/temporalio/sig/temporalio/converters/storage_driver_workflow_info.rbs b/temporalio/sig/temporalio/converters/storage_driver_workflow_info.rbs
new file mode 100644
index 00000000..b27139bc
--- /dev/null
+++ b/temporalio/sig/temporalio/converters/storage_driver_workflow_info.rbs
@@ -0,0 +1,17 @@
+module Temporalio
+  module Converters
+    class StorageDriverWorkflowInfo
+      attr_reader namespace: String
+      attr_reader id: String?
+      attr_reader run_id: String?
+      attr_reader type: String?
+
+      def initialize: (
+        namespace: String,
+        ?id: String?,
+        ?run_id: String?,
+        ?type: String?
+      ) -> void
+    end
+  end
+end
diff --git a/temporalio/sig/temporalio/error.rbs b/temporalio/sig/temporalio/error.rbs
index 72675e9b..1cd4771b 100644
--- a/temporalio/sig/temporalio/error.rbs
+++ b/temporalio/sig/temporalio/error.rbs
@@ -49,6 +49,10 @@ module Temporalio
       def initialize: (Activity::CancellationDetails details) -> void
     end
 
+    class ExternalStorageNotConfiguredError < Error
+      def initialize: (?String message) -> void
+    end
+
     class RPCError < Error
       attr_reader code: Code::enum
 
diff --git a/temporalio/sig/temporalio/internal/external_storage_references.rbs b/temporalio/sig/temporalio/internal/external_storage_references.rbs
new file mode 100644
index 00000000..b4cf9312
--- /dev/null
+++ b/temporalio/sig/temporalio/internal/external_storage_references.rbs
@@ -0,0 +1,18 @@
+module Temporalio
+  module Internal
+    module ExternalStorageReferences
+      ENCODING: String
+      MESSAGE_TYPE: String
+
+      def self.reference?: (Api::Common::V1::Payload payload) -> bool
+      def self.parse_reference: (
+        Api::Common::V1::Payload payload
+      ) -> Api::Sdk::V1::ExternalStorageReference?
+      def self.create_reference_payload: (
+        driver_name: String,
+        claim_data: Hash[String, String],
+        original_size_bytes: Integer
+      ) -> Api::Common::V1::Payload
+    end
+  end
+end
diff --git a/temporalio/test/external_storage_test.rb b/temporalio/test/external_storage_test.rb
new file mode 100644
index 00000000..19397a4b
--- /dev/null
+++ b/temporalio/test/external_storage_test.rb
@@ -0,0 +1,267 @@
+# frozen_string_literal: true
+
+require 'base64'
+require 'in_memory_storage_driver'
+require 'securerandom'
+require 'temporalio/cancellation'
+require 'temporalio/converters/external_storage'
+require 'temporalio/converters/storage_driver_activity_info'
+require 'temporalio/converters/storage_driver_claim'
+require 'temporalio/converters/storage_driver_retrieve_context'
+require 'temporalio/converters/storage_driver_select_context'
+require 'temporalio/converters/storage_driver_store_context'
+require 'temporalio/converters/storage_driver_workflow_info'
+require 'temporalio/error'
+require 'temporalio/internal/external_storage_references'
+require 'test'
+
+class ExternalStorageTest < Test
+  References = Temporalio::Internal::ExternalStorageReferences
+
+  def test_reference_payload_round_trips
+    payload = References.create_reference_payload(
+      driver_name: 'mem', claim_data: { 'key' => 'k' }, original_size_bytes: 385
+    )
+
+    assert References.reference?(payload)
+    assert_equal 'json/protobuf', payload.metadata['encoding']
+    assert_equal 'temporal.api.sdk.v1.ExternalStorageReference', payload.metadata['messageType']
+    assert_equal 385, payload.external_payloads.first.size_bytes
+
+    reference = References.parse_reference(payload)
+    assert_equal 'mem', reference.driver_name
+    assert_equal({ 'key' => 'k' }, reference.claim_data.to_h)
+  end
+
+  def test_ordinary_payload_is_not_a_reference
+    payload = Temporalio::Api::Common::V1::Payload.new(metadata: { 'encoding' => 'json/plain' }, data: '{}')
+
+    refute References.reference?(payload)
+    assert_nil References.parse_reference(payload)
+  end
+
+  def test_reference_with_unknown_field_parses
+    # A newer SDK may add fields to the reference before this one knows about them.
+    payload = Temporalio::Api::Common::V1::Payload.new(
+      metadata: { 'encoding' => References::ENCODING, 'messageType' => References::MESSAGE_TYPE },
+      data: '{"driverName":"mem","claimData":{"k":"v"},"fieldFromANewerSdk":42}'
+    )
+
+    reference = References.parse_reference(payload)
+
+    assert_equal 'mem', reference.driver_name
+    assert_equal({ 'k' => 'v' }, reference.claim_data.to_h)
+  end
+
+  def test_reference_from_another_sdk_parses
+    # Verbatim from the Go SDK's TestClaimDeserialization_OtherSdk_ProtoJSON fixture: compact and differently ordered
+    # JSON, which is what another SDK actually puts on the wire.
+    data = Base64.decode64(
+      'eyJjbGFpbURhdGEiOnsiYnVja2V0IjoidGVzdC1idWNrZXQiLCJoYXNoX2FsZ29yaXRobSI6InNoYTI1NiIsImhhc2hfdmFsdWUi' \
+      'OiI2Y2EyMmMzNDU2MGNmMzVhYzI0NDI3ZGM3NjE5YzlhYjQ3MmE4MmNmMThmMjg2ZjI3ODcxNjQ5YTJiNTYwOGM4Iiwia2V5Ijoi' \
+      'djAvbnMvZGVmYXVsdC93dC9MYXJnZUlPV29ya2Zsb3cvd2kvZjFkMmE0YWMtZjhjYi00NWQzLTkwOGMtOTNhMGYzM2FiMjQ1L3Jp' \
+      'L251bGwvZC9zaGEyNTYvNmNhMjJjMzQ1NjBjZjM1YWMyNDQyN2RjNzYxOWM5YWI0NzJhODJjZjE4ZjI4NmYyNzg3MTY0OWEyYjU2' \
+      'MDhjOCJ9LCJkcml2ZXJOYW1lIjoiYXdzLnMzZHJpdmVyIn0='
+    )
+    payload = Temporalio::Api::Common::V1::Payload.new(
+      metadata: { 'encoding' => 'json/protobuf', 'messageType' => References::MESSAGE_TYPE },
+      data:
+    )
+
+    reference = References.parse_reference(payload)
+
+    assert_equal 'aws.s3driver', reference.driver_name
+    assert_equal 'test-bucket', reference.claim_data['bucket']
+    assert_equal 4, reference.claim_data.size
+  end
+
+  def test_single_driver_synthesizes_selector
+    driver = InMemoryStorageDriver.new
+    storage = Temporalio::Converters::ExternalStorage.new(drivers: [driver])
+
+    assert_equal 256 * 1024, storage.payload_size_threshold
+    assert_same driver, storage.driver_selector.call(select_context, payload('x'))
+    assert_same driver, storage.driver('mem')
+    assert_nil storage.driver('missing')
+  end
+
+  def test_mutating_the_given_driver_array_does_not_affect_storage
+    driver = InMemoryStorageDriver.new
+    drivers = [driver]
+    storage = Temporalio::Converters::ExternalStorage.new(drivers:)
+
+    drivers << InMemoryStorageDriver.new(name: 'added')
+    drivers.clear
+
+    assert_same driver, storage.drivers.first
+    assert_same driver, storage.driver_selector.call(select_context, payload('x'))
+    assert_nil storage.driver('added')
+  end
+
+  def test_drivers_is_frozen
+    storage = Temporalio::Converters::ExternalStorage.new(drivers: [InMemoryStorageDriver.new])
+
+    assert_predicate storage.drivers, :frozen?
+    assert_raises(FrozenError) { storage.drivers << InMemoryStorageDriver.new(name: 'added') }
+  end
+
+  def test_no_drivers_rejected
+    err = assert_raises(ArgumentError) { Temporalio::Converters::ExternalStorage.new(drivers: []) }
+    assert_includes err.message, 'At least one driver'
+  end
+
+  def test_empty_driver_name_rejected
+    err = assert_raises(ArgumentError) do
+      Temporalio::Converters::ExternalStorage.new(drivers: [InMemoryStorageDriver.new(name: '')])
+    end
+    assert_includes err.message, 'must be a non-empty String'
+  end
+
+  def test_symbol_driver_name_rejected
+    # A Symbol passes a bare empty? check but cannot be looked up by String or written to the proto.
+    err = assert_raises(ArgumentError) do
+      Temporalio::Converters::ExternalStorage.new(drivers: [InMemoryStorageDriver.new(name: :mem)])
+    end
+    assert_includes err.message, 'must be a non-empty String'
+  end
+
+  def test_duplicate_driver_names_rejected
+    err = assert_raises(ArgumentError) do
+      Temporalio::Converters::ExternalStorage.new(
+        drivers: [InMemoryStorageDriver.new(name: 'dup'), InMemoryStorageDriver.new(name: 'dup')],
+        driver_selector: ->(_context, _payload) {}
+      )
+    end
+    assert_includes err.message, "name 'dup'"
+  end
+
+  def test_multiple_drivers_require_selector
+    err = assert_raises(ArgumentError) do
+      Temporalio::Converters::ExternalStorage.new(
+        drivers: [InMemoryStorageDriver.new(name: 'a'), InMemoryStorageDriver.new(name: 'b')]
+      )
+    end
+    assert_includes err.message, 'driver_selector is required'
+  end
+
+  def test_negative_threshold_rejected
+    err = assert_raises(ArgumentError) do
+      Temporalio::Converters::ExternalStorage.new(drivers: [InMemoryStorageDriver.new], payload_size_threshold: -1)
+    end
+    assert_includes err.message, 'non-negative Integer'
+  end
+
+  def test_non_integer_threshold_rejected
+    block = proc do
+      [nil, '256', 256.0].each do |value|
+        err = assert_raises(ArgumentError) do
+          Temporalio::Converters::ExternalStorage.new(
+            drivers: [InMemoryStorageDriver.new], payload_size_threshold: value # steep:ignore
+          )
+        end
+        assert_includes err.message, 'non-negative Integer'
+      end
+    end
+    # Passing the wrong type is the point of this test, so the runtime signature check must not record it.
+    if defined?(SigApplicator)
+      SigApplicator.suppress_errors { block.call }
+    else
+      block.call
+    end
+  end
+
+  def test_zero_threshold_allowed
+    # Zero offloads every payload rather than disabling offload.
+    storage = Temporalio::Converters::ExternalStorage.new(
+      drivers: [InMemoryStorageDriver.new], payload_size_threshold: 0
+    )
+
+    assert_equal 0, storage.payload_size_threshold
+  end
+
+  def test_selector_may_decline
+    storage = Temporalio::Converters::ExternalStorage.new(
+      drivers: [InMemoryStorageDriver.new], driver_selector: ->(_context, _payload) {}
+    )
+
+    assert_nil storage.driver_selector.call(select_context, payload('x'))
+  end
+
+  def test_driver_store_and_retrieve_round_trip
+    driver = InMemoryStorageDriver.new
+    payloads = [payload('one'), payload('two')]
+
+    claims = driver.store(store_context, payloads)
+    restored = driver.retrieve(Temporalio::Converters::StorageDriverRetrieveContext.new(cancellation:), claims)
+
+    assert_equal 2, claims.size
+    assert_equal payloads, restored
+  end
+
+  def test_claims_with_equal_content_are_equal
+    first = Temporalio::Converters::StorageDriverClaim.new(claim_data: { 'a' => '1', 'b' => '2' })
+    second = Temporalio::Converters::StorageDriverClaim.new(claim_data: { 'b' => '2', 'a' => '1' })
+
+    assert_equal first, second
+    assert_equal first.hash, second.hash
+    assert first.eql?(second)
+  end
+
+  def test_mutating_the_given_claim_data_does_not_affect_the_claim
+    value = +'my-bucket'
+    claim = Temporalio::Converters::StorageDriverClaim.new(claim_data: { 'bucket' => value })
+    hash_before = claim.hash
+
+    value << '-tampered'
+
+    assert_equal 'my-bucket', claim.claim_data.fetch('bucket')
+    assert_equal hash_before, claim.hash
+    assert_predicate claim.claim_data, :frozen?
+    assert_raises(FrozenError) { claim.claim_data.fetch('bucket') << 'x' }
+  end
+
+  def test_claims_with_different_content_are_not_equal
+    claim = Temporalio::Converters::StorageDriverClaim.new(claim_data: { 'key' => 'k' })
+
+    refute_equal claim, Temporalio::Converters::StorageDriverClaim.new(claim_data: { 'key' => 'other' })
+    refute_equal claim, Temporalio::Converters::StorageDriverClaim.new(claim_data: { 'another' => 'k' })
+    refute_equal claim, Temporalio::Converters::StorageDriverClaim.new(claim_data: { 'key' => 'k', 'x' => 'y' })
+    refute_equal claim, 'not a claim'
+  end
+
+  def test_target_info_shapes
+    workflow = Temporalio::Converters::StorageDriverWorkflowInfo.new(namespace: 'ns', id: 'wf', run_id: 'r',
+                                                                     type: 'T')
+    activity = Temporalio::Converters::StorageDriverActivityInfo.new(namespace: 'ns')
+
+    assert_equal %w[ns wf r T], [workflow.namespace, workflow.id, workflow.run_id, workflow.type]
+    assert_equal 'ns', activity.namespace
+    assert_nil activity.id
+    assert_nil activity.run_id
+    assert_nil activity.type
+  end
+
+  def test_not_configured_error_names_its_code
+    err = Temporalio::Error::ExternalStorageNotConfiguredError.new
+
+    assert_includes err.message, 'TMPRL1105'
+  end
+
+  private
+
+  def cancellation
+    Temporalio::Cancellation.new
+  end
+
+  def select_context
+    Temporalio::Converters::StorageDriverSelectContext.new(target: nil, cancellation:)
+  end
+
+  def store_context
+    Temporalio::Converters::StorageDriverStoreContext.new(target: nil, cancellation:)
+  end
+
+  def payload(data)
+    Temporalio::Api::Common::V1::Payload.new(metadata: { 'encoding' => 'json/plain' }, data:)
+  end
+end
diff --git a/temporalio/test/in_memory_storage_driver.rb b/temporalio/test/in_memory_storage_driver.rb
new file mode 100644
index 00000000..1aacef35
--- /dev/null
+++ b/temporalio/test/in_memory_storage_driver.rb
@@ -0,0 +1,54 @@
+# frozen_string_literal: true
+
+require 'securerandom'
+require 'temporalio/api'
+require 'temporalio/converters/storage_driver'
+require 'temporalio/converters/storage_driver_claim'
+
+# Storage driver that keeps payloads in a hash, recording what it was asked to do so tests can assert on batching
+# and targets.
+#
+# One instance is shared across whatever concurrency a test drives it with, so all of its state is guarded and the
+# readers hand back snapshots rather than the live collections.
+class InMemoryStorageDriver < Temporalio::Converters::StorageDriver
+  attr_reader :name, :type
+
+  def initialize(name: 'mem', type: 'test.memdriver')
+    super()
+    @name = name
+    @type = type
+    @mutex = Mutex.new
+    @stored = {}
+    @store_calls = []
+    @retrieve_calls = []
+  end
+
+  def stored
+    @mutex.synchronize { @stored.dup }
+  end
+
+  def store_calls
+    @mutex.synchronize { @store_calls.dup }
+  end
+
+  def retrieve_calls
+    @mutex.synchronize { @retrieve_calls.dup }
+  end
+
+  def store(context, payloads)
+    @mutex.synchronize { @store_calls << [context, payloads.size] }
+    payloads.map do |payload|
+      key = SecureRandom.uuid
+      @mutex.synchronize { @stored[key] = payload.to_proto }
+      Temporalio::Converters::StorageDriverClaim.new(claim_data: { 'key' => key })
+    end
+  end
+
+  def retrieve(_context, claims)
+    @mutex.synchronize { @retrieve_calls << claims.size }
+    claims.map do |claim|
+      bytes = @mutex.synchronize { @stored.fetch(claim.claim_data.fetch('key')) }
+      Temporalio::Api::Common::V1::Payload.decode(bytes)
+    end
+  end
+end
diff --git a/temporalio/test/sig/external_storage_test.rbs b/temporalio/test/sig/external_storage_test.rbs
new file mode 100644
index 00000000..c6dea28f
--- /dev/null
+++ b/temporalio/test/sig/external_storage_test.rbs
@@ -0,0 +1,6 @@
+class ExternalStorageTest < Test
+  def cancellation: -> Temporalio::Cancellation
+  def select_context: -> Temporalio::Converters::StorageDriverSelectContext
+  def store_context: -> Temporalio::Converters::StorageDriverStoreContext
+  def payload: (String data) -> Temporalio::Api::Common::V1::Payload
+end