diff --git a/gapic-generator-ads/lib/gapic/generators/ads_generator.rb b/gapic-generator-ads/lib/gapic/generators/ads_generator.rb
index d17dc8d6f..0f5d4e2db 100644
--- a/gapic-generator-ads/lib/gapic/generators/ads_generator.rb
+++ b/gapic-generator-ads/lib/gapic/generators/ads_generator.rb
@@ -63,6 +63,7 @@ def generate
             files << g("service/credentials", "lib/#{service.credentials_file_path}",  service: service)
             files << g("service/paths",       "lib/#{service.paths_file_path}",        service: service) if service.paths?
             files << g("service/operations",  "lib/#{service.operations_file_path}",   service: service) if service.lro?
+            files << g("service/resumable_upload_stub", "lib/#{service.resumable_upload_stub_file_path}", service: service) if service.resumable_upload?
           end
         end
 
diff --git a/gapic-generator-cloud/templates/cloud/service/client/method/def/_upload_error_handler.text.erb b/gapic-generator-cloud/templates/cloud/service/client/method/def/_upload_error_handler.text.erb
new file mode 100644
index 000000000..8e0c763cc
--- /dev/null
+++ b/gapic-generator-cloud/templates/cloud/service/client/method/def/_upload_error_handler.text.erb
@@ -0,0 +1,2 @@
+<%- assert_locals method -%>
+->(e) { ::Google::Cloud::Error.from_error e }
diff --git a/gapic-generator-cloud/templates/cloud/service/rest/client/method/def/_upload_error_handler.text.erb b/gapic-generator-cloud/templates/cloud/service/rest/client/method/def/_upload_error_handler.text.erb
new file mode 100644
index 000000000..8e0c763cc
--- /dev/null
+++ b/gapic-generator-cloud/templates/cloud/service/rest/client/method/def/_upload_error_handler.text.erb
@@ -0,0 +1,2 @@
+<%- assert_locals method -%>
+->(e) { ::Google::Cloud::Error.from_error e }
diff --git a/gapic-generator/lib/gapic/generators/default_generator.rb b/gapic-generator/lib/gapic/generators/default_generator.rb
index c1612a78b..23eb71072 100644
--- a/gapic-generator/lib/gapic/generators/default_generator.rb
+++ b/gapic-generator/lib/gapic/generators/default_generator.rb
@@ -92,10 +92,16 @@ def generate gem_presenter: nil
             # Rest-only `service.stub` file
             files << g("service/rest/service_stub",      "lib/#{service.rest.service_stub_file_path}",       service: service) if should_generate_rest
 
+            # Resumable upload stub, shared by both transports because uploads always travel over REST
+            files << g("service/resumable_upload_stub",  "lib/#{service.resumable_upload_stub_file_path}",   service: service) if service.resumable_upload?
+
             # Unit tests for `client.rb`
             files << g("service/test/client",            "test/#{service.test_client_file_path}",            service: service) if should_generate_grpc
             files << g("service/rest/test/client",       "test/#{service.rest.test_client_file_path}",       service: service) if should_generate_rest
 
+            # Unit tests for resumable upload RPCs, which the client tests above skip
+            files << g("service/test/resumable_upload",  "test/#{service.test_resumable_upload_file_path}",  service: service) if service.resumable_upload?
+
             # Unit tests for `paths.rb`
             files << g("service/test/client_paths",      "test/#{service.test_paths_file_path}",             service: service) if service.paths? && should_generate_grpc
 
diff --git a/gapic-generator/lib/gapic/model/method/resumable_upload.rb b/gapic-generator/lib/gapic/model/method/resumable_upload.rb
new file mode 100644
index 000000000..f56df3d1f
--- /dev/null
+++ b/gapic-generator/lib/gapic/model/method/resumable_upload.rb
@@ -0,0 +1,143 @@
+# frozen_string_literal: true
+
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+require "gapic/model/model_error"
+
+module Gapic
+  module Model
+    module Method
+      ##
+      # Resumable upload method-level model.
+      #
+      # A resumable upload RPC does not send its payload in the initiation request. The request it
+      # describes only creates an upload session; the bytes travel afterwards, over REST, in chunks
+      # addressed to a URL the server hands back. Generated clients therefore return an upload handle
+      # from such a method rather than a response message.
+      #
+      # Until the upload annotation is published, the set of such RPCs is carried here as a table, and
+      # so is the URL prefix each one's initiation request is sent under. When the annotation lands,
+      # {.url_prefix_for} keeps the table and detection moves to `http.media_upload.enabled`.
+      #
+      # @!attribute [r] url_prefix
+      #   @return [String] The path prefix prepended to the transcoded initiation URL, without
+      #     surrounding slashes, e.g. `resumable/upload`.
+      #
+      class ResumableUpload
+        # @return [String]
+        attr_reader :url_prefix
+
+        ##
+        # @param url_prefix [String] The upload URL prefix for the matched RPC.
+        #
+        def initialize url_prefix
+          @url_prefix = url_prefix
+        end
+
+        ##
+        # Exact matches, keyed by the full gRPC name of the RPC.
+        #
+        EXACT_PREFIXES = {
+          "google.showcase.v1beta1.ResumableUploadService.UploadMedia" => "resumable/upload"
+        }.freeze
+
+        ##
+        # Version-family matches, for protos that are republished under a new version regularly.
+        # Anchored on the left at the package and on the right at the service and method, with the
+        # intervening segments (e.g. `.services.`) unconstrained.
+        #
+        VERSIONED_PREFIXES = [
+          {
+            left:   /\Agoogle\.ads\.googleads\.v[0-9_]+\./,
+            right:  ".YouTubeVideoUploadService.CreateYouTubeVideoUpload",
+            prefix: "resumable/upload"
+          }
+        ].freeze
+
+        class << self
+          ##
+          # Inspects a method and returns its resumable upload model, or `nil` if it does not perform
+          # resumable uploads.
+          #
+          # @param method [Gapic::Presenters::MethodPresenter]
+          #
+          # @raise [Gapic::Model::ModelError] if the method is a resumable upload RPC that the
+          #   generator cannot generate an upload surface for.
+          #
+          # @return [Gapic::Model::Method::ResumableUpload, nil]
+          #
+          def create method
+            prefix = url_prefix_for method.grpc_full_name
+            return nil if prefix.nil?
+            validate! method
+            new prefix
+          end
+
+          ##
+          # The upload URL prefix for an RPC, or `nil` if the RPC does not perform resumable uploads.
+          #
+          # @param full_name [String] The full gRPC name of the RPC,
+          #   e.g. `google.showcase.v1beta1.ResumableUploadService.UploadMedia`.
+          #
+          # @return [String, nil]
+          #
+          def url_prefix_for full_name
+            EXACT_PREFIXES[full_name] ||
+              VERSIONED_PREFIXES.find do |match|
+                match[:left].match?(full_name) && full_name.end_with?(match[:right])
+              end&.fetch(:prefix)
+          end
+
+          ##
+          # Verifies that an upload surface can be generated for the given method. A resumable upload
+          # is a single unary POST that carries a body, and anything else in the table is a
+          # misconfiguration that must fail the build rather than generate code that cannot work.
+          #
+          # @param method [Gapic::Presenters::MethodPresenter]
+          #
+          # @raise [Gapic::Model::ModelError]
+          #
+          # @return [void]
+          #
+          def validate! method
+            reason = unsupported_reason method
+            return if reason.nil?
+            raise ModelError, "The method #{method.grpc_full_name} performs resumable uploads, " \
+                              "which the generator supports only for #{reason}."
+          end
+
+          private
+
+          ##
+          # @param method [Gapic::Presenters::MethodPresenter]
+          # @return [String, nil] What the method would have had to be, or `nil` if it is supported.
+          #
+          def unsupported_reason method
+            return "non-streaming methods" if method.client_streaming? || method.server_streaming?
+            return "non-paginated methods" if method.paged?
+            return "methods that are not long-running operations" if method.lro? || method.nonstandard_lro?
+
+            binding = method.http_bindings.first
+            return "methods with an HTTP binding" if binding.nil?
+            return "methods bound to POST" unless binding.verb == :post
+            return "methods whose HTTP binding has a body" unless binding.body?
+
+            nil
+          end
+        end
+      end
+    end
+  end
+end
diff --git a/gapic-generator/lib/gapic/presenters/method_presenter.rb b/gapic-generator/lib/gapic/presenters/method_presenter.rb
index 10375cd2b..1af42aa6b 100644
--- a/gapic-generator/lib/gapic/presenters/method_presenter.rb
+++ b/gapic-generator/lib/gapic/presenters/method_presenter.rb
@@ -17,6 +17,7 @@
 require "active_support/inflector"
 require "gapic/ruby_info"
 require "gapic/helpers/namespace_helper"
+require "gapic/model/method/resumable_upload"
 
 module Gapic
   module Presenters
@@ -71,6 +72,10 @@ def initialize service_presenter, api, method
         @lro = Gapic::Model::Method.parse_lro @method, @api
 
         @rest = MethodRestPresenter.new self, @api
+
+        # Built last: detection is cheap but its validation reads the LRO model, the HTTP bindings
+        # and the pagination check, all of which have to exist first.
+        @resumable_upload = Gapic::Model::Method::ResumableUpload.create self
       end
 
       ##
@@ -279,6 +284,36 @@ def nonstandard_lro_client
         service.nonstandard_lros.find { |model| model.service == @lro.service_full_name }
       end
 
+      ##
+      # Whether this method performs a resumable upload. Such a method returns an upload handle
+      # rather than a response, and its payload travels over REST in chunks after the request this
+      # method describes has created the upload session.
+      #
+      # @return [Boolean]
+      #
+      def resumable_upload?
+        !@resumable_upload.nil?
+      end
+
+      ##
+      # The path prefix prepended to this method's transcoded initiation URL, without surrounding
+      # slashes, e.g. `resumable/upload`. `nil` unless this method performs a resumable upload.
+      #
+      # @return [String, nil]
+      #
+      def upload_url_prefix
+        @resumable_upload&.url_prefix
+      end
+
+      ##
+      # The name of the constant the generated upload stub holds this method's URL prefix in.
+      #
+      # @return [String]
+      #
+      def upload_url_prefix_const_name
+        "#{name.upcase}_URL_PREFIX"
+      end
+
       def client_streaming?
         @method.client_streaming
       end
diff --git a/gapic-generator/lib/gapic/presenters/service_presenter.rb b/gapic-generator/lib/gapic/presenters/service_presenter.rb
index ebea2bca9..c9928e632 100644
--- a/gapic-generator/lib/gapic/presenters/service_presenter.rb
+++ b/gapic-generator/lib/gapic/presenters/service_presenter.rb
@@ -422,6 +422,15 @@ def test_client_operations_file_path
         service_file_path.sub ".rb", "_operations_test.rb"
       end
 
+      ##
+      # Path of the generated tests covering this service's resumable upload RPCs. Those RPCs are
+      # excluded from the ordinary client tests, which assume a call returns a response.
+      #
+      # @return [String]
+      def test_resumable_upload_file_path
+        service_file_path.sub ".rb", "_resumable_upload_test.rb"
+      end
+
       def stub_name
         "#{ActiveSupport::Inflector.underscore name}_stub"
       end
@@ -489,6 +498,62 @@ def lro_service
         ServicePresenter.new @gem_presenter, @api, lro.services.first, parent_service: self unless lro.nil?
       end
 
+      ##
+      # Whether any of this service's RPCs perform resumable uploads, and therefore whether an upload
+      # stub has to be generated for it and built by its clients.
+      #
+      # @return [Boolean]
+      def resumable_upload?
+        methods.any?(&:resumable_upload?)
+      end
+
+      ##
+      # Presenters for the RPCs of this service that perform resumable uploads.
+      #
+      # @return [Enumerable<Gapic::Presenters::MethodPresenter>]
+      def resumable_upload_methods
+        methods.select(&:resumable_upload?)
+      end
+
+      ##
+      # The class name of the generated upload stub. One per service, shared by both transports, and
+      # deliberately not nested under `Rest::`: the gRPC client builds it too, because the upload
+      # itself always travels over REST.
+      #
+      # @return [String]
+      def resumable_upload_stub_name
+        "ResumableUploadStub"
+      end
+
+      # @return [String]
+      def resumable_upload_stub_name_full
+        fix_namespace @api, "#{service_name_full}::#{resumable_upload_stub_name}"
+      end
+
+      # @return [String]
+      def resumable_upload_stub_require
+        ruby_file_path @api, resumable_upload_stub_name_full
+      end
+
+      # @return [String]
+      def resumable_upload_stub_file_path
+        "#{resumable_upload_stub_require}.rb"
+      end
+
+      # @return [String]
+      def resumable_upload_stub_file_name
+        resumable_upload_stub_file_path.split("/").last
+      end
+
+      ##
+      # An instance variable name used for the generated upload stub. The clients keep the stub here
+      # and expose no reader for it.
+      #
+      # @return [String]
+      def resumable_upload_stub_ivar
+        "@resumable_upload_stub"
+      end
+
       def config_channel_args
         { "grpc.service_config_disable_resolution" => 1 }
       end
diff --git a/gapic-generator/lib/gapic/presenters/service_rest_presenter.rb b/gapic-generator/lib/gapic/presenters/service_rest_presenter.rb
index 133e2f111..9cfa78f5f 100644
--- a/gapic-generator/lib/gapic/presenters/service_rest_presenter.rb
+++ b/gapic-generator/lib/gapic/presenters/service_rest_presenter.rb
@@ -357,6 +357,18 @@ def methods
         main_service.methods.select(&:can_generate_rest?)
       end
 
+      ##
+      # Presenters for methods that the REST service stub carries an implementation for. An upload
+      # RPC has no ordinary REST path: the REST client delegates to the upload handle exactly as the
+      # gRPC client does, so a plain call method and transcoder here would be dead code that also
+      # happens to be wrong — a non-resumable POST of the whole payload.
+      #
+      # @return [Enumerable<Gapic::Presenters::MethodPresenter>]
+      #
+      def service_stub_methods
+        methods.reject(&:resumable_upload?)
+      end
+
       ##
       # Require string for the helpers file
       #
diff --git a/gapic-generator/lib/gapic/presenters/snippet/response_handling_presenters.rb b/gapic-generator/lib/gapic/presenters/snippet/response_handling_presenters.rb
index b8f150c6c..78779beff 100644
--- a/gapic-generator/lib/gapic/presenters/snippet/response_handling_presenters.rb
+++ b/gapic-generator/lib/gapic/presenters/snippet/response_handling_presenters.rb
@@ -80,6 +80,66 @@ def initialize proto, _json, response_type:, phase1:
         attr_reader :response_name
       end
 
+      ##
+      # Presentation information about resumable upload response handling.
+      #
+      # A resumable upload RPC returns a {::Gapic::ResumableUpload} handle rather than a
+      # response message, so the snippet names the call result `upload` and goes on to
+      # start the upload, which is what actually produces the response message.
+      #
+      class ResumableUploadResponseHandlingPresenter
+        include ResponseHandlingPresenterCommon
+
+        ##
+        # Create a resumable upload response handling presenter
+        #
+        # @param proto [Google::Cloud::Tools::SnippetGen::ConfigLanguage::V1::Snippet::SimpleResponseHandling]
+        #     The protobuf representation
+        # @param json [String]
+        #     The JSON representation
+        # @param response_type [String] The fully qualified response message class
+        # @param phase1 [Boolean] True if this is a phase 1 snippet without config
+        #
+        def initialize proto, _json, response_type:, phase1:
+          @response_name = phase1 ? "upload" : compute_response_name(proto, phase1)
+          @render_lines = phase1 ? upload_lines(response_type) : []
+          @render = @render_lines.join "\n"
+        end
+
+        ##
+        # The lines of rendered code
+        # @return [Array<String>]
+        #
+        attr_reader :render_lines
+
+        ##
+        # The rendered code as a single string, possibly with line breaks
+        # @return [String]
+        #
+        attr_reader :render
+
+        ##
+        # The name of the response variable, or nil for no response handling
+        # @return [String,nil]
+        #
+        attr_reader :response_name
+
+        private
+
+        def upload_lines response_type
+          [
+            "# The returned object is a handle for a resumable upload. Nothing has been",
+            "# uploaded yet, and the timeout and retry policy of the call above cover only",
+            "# the request that creates the upload session, not the upload as a whole.",
+            "stream = File.open \"input.bin\", \"rb\"",
+            "result = #{@response_name}.start stream: stream, content_type: \"application/octet-stream\"",
+            "",
+            "# The returned object is of type #{response_type}.",
+            "p result"
+          ]
+        end
+      end
+
       ##
       # Presentation information about LRO response handling
       #
diff --git a/gapic-generator/lib/gapic/presenters/snippet_presenter.rb b/gapic-generator/lib/gapic/presenters/snippet_presenter.rb
index 850139308..83b861e35 100644
--- a/gapic-generator/lib/gapic/presenters/snippet_presenter.rb
+++ b/gapic-generator/lib/gapic/presenters/snippet_presenter.rb
@@ -63,6 +63,8 @@ def response_kind
           :paged
         elsif @method_presenter.lro?
           :lro
+        elsif @method_presenter.resumable_upload?
+          :resumable_upload
         else
           :simple
         end
@@ -244,13 +246,12 @@ def build_response_handling_presenter call_proto, call_json
           LroResponseHandlingPresenter.new call_proto&.lro_handling,
                                            call_json&.fetch("lroHandling", nil),
                                            phase1: phase1
+        when :resumable_upload
+          ResumableUploadResponseHandlingPresenter.new call_proto&.response_handling,
+                                                       call_json&.fetch("responseHandling", nil),
+                                                       response_type: return_type, phase1: phase1
         when :streaming
-          response_name = phase1 ? "output" : call_proto&.server_stream_name
-          response_name = nil if response_name == ""
-          StreamingResponseHandlingPresenter.new call_proto&.response_handling,
-                                                 call_json&.fetch("responseHandling", nil),
-                                                 response_name: response_name, base_response_type: base_response_type,
-                                                 phase1: phase1
+          build_streaming_response_handling_presenter call_proto, call_json, phase1
         else
           SimpleResponseHandlingPresenter.new call_proto&.response_handling,
                                               call_json&.fetch("responseHandling", nil),
@@ -258,6 +259,15 @@ def build_response_handling_presenter call_proto, call_json
         end
       end
 
+      def build_streaming_response_handling_presenter call_proto, call_json, phase1
+        response_name = phase1 ? "output" : call_proto&.server_stream_name
+        response_name = nil if response_name == ""
+        StreamingResponseHandlingPresenter.new call_proto&.response_handling,
+                                               call_json&.fetch("responseHandling", nil),
+                                               response_name: response_name, base_response_type: base_response_type,
+                                               phase1: phase1
+      end
+
       def build_client_call_presenter call_proto, call_json, request_name, response_name
         phase1 = !config?
         if response_kind == :paged
diff --git a/gapic-generator/templates/default/lib/_service.text.erb b/gapic-generator/templates/default/lib/_service.text.erb
index 2de619a6b..f83b456d0 100644
--- a/gapic-generator/templates/default/lib/_service.text.erb
+++ b/gapic-generator/templates/default/lib/_service.text.erb
@@ -15,6 +15,9 @@ require "<%= service.credentials_require %>"
 <%- if service.paths? -%>
 require "<%= service.paths_require %>"
 <%- end -%>
+<%- if service.resumable_upload? -%>
+require "<%= service.resumable_upload_stub_require %>"
+<%- end -%>
 <%- if service.generate_grpc_clients? -%>
 <%- if service.lro? -%>
 require "<%= service.operations_require %>"
diff --git a/gapic-generator/templates/default/lib/rest/_rest.text.erb b/gapic-generator/templates/default/lib/rest/_rest.text.erb
index da0d8bb87..70dc25d3c 100644
--- a/gapic-generator/templates/default/lib/rest/_rest.text.erb
+++ b/gapic-generator/templates/default/lib/rest/_rest.text.erb
@@ -15,6 +15,9 @@ require "<%= service.credentials_require %>"
 <%- if service.paths? -%>
 require "<%= service.paths_require %>"
 <%- end -%>
+<%- if service.resumable_upload? -%>
+require "<%= service.resumable_upload_stub_require %>"
+<%- end -%>
 <%- if service.rest.lro? -%>
 require "<%= service.rest.operations_require %>"
 <%- end -%>
diff --git a/gapic-generator/templates/default/service/client/_client.text.erb b/gapic-generator/templates/default/service/client/_client.text.erb
index 91d0ef532..dff69766b 100644
--- a/gapic-generator/templates/default/service/client/_client.text.erb
+++ b/gapic-generator/templates/default/service/client/_client.text.erb
@@ -20,6 +20,14 @@ require "securerandom"
 <%= indent service.doc_description(transport: :grpc), "# " %>
 #
 <%- end -%>
+<%- if service.resumable_upload? -%>
+<%- upload_names = service.resumable_upload_methods.map { |upload_method| "`#{upload_method.name}`" }.join ", " -%>
+<%- upload_many = service.resumable_upload_methods.length > 1 -%>
+# <%= upload_names %> <%= upload_many ? "perform resumable uploads: each returns" : "performs a resumable upload: it returns" %> a {::Gapic::ResumableUpload} handle
+# instead of a response, and the per-call `timeout` and retry policy cover only the request that
+# creates the upload session, not the upload itself.
+#
+<%- end -%>
 class <%= service.client_name %>
   # @private
   API_VERSION = "<%= service.api_version %>".freeze
@@ -140,6 +148,16 @@ class <%= service.client_name %>
       config.universe_domain = @config.universe_domain
     end
 
+    <%- end -%>
+    <%- if service.resumable_upload? -%>
+    <%= service.resumable_upload_stub_ivar %> = <%= service.resumable_upload_stub_name_full %>.new(
+      endpoint: @config.endpoint,
+      endpoint_template: DEFAULT_ENDPOINT_TEMPLATE,
+      universe_domain: @config.universe_domain,
+      credentials: credentials,
+      logger: @config.logger
+    )
+
     <%- end -%>
     @<%= service.stub_name %> = ::Gapic::ServiceStub.new(
       <%= service.proto_service_stub_name_full %>,
diff --git a/gapic-generator/templates/default/service/client/_config.text.erb b/gapic-generator/templates/default/service/client/_config.text.erb
index 1c171a035..49fff6e15 100644
--- a/gapic-generator/templates/default/service/client/_config.text.erb
+++ b/gapic-generator/templates/default/service/client/_config.text.erb
@@ -190,6 +190,14 @@ class Configuration
   <%- method_service.methods.each do |method| -%>
     ##
     # RPC-specific configuration for `<%= method.name %>`
+  <%- if method.resumable_upload? -%>
+    #
+    # `<%= method.name %>` performs a resumable upload, so the `timeout` and `retry_policy`
+    # configured here cover the request that creates the upload session only, not the upload
+    # itself. The whole-upload budget is the `upload_timeout:` argument of
+    # {::Gapic::ResumableUpload#start} and {::Gapic::ResumableUpload#resume}.
+    #
+  <%- end -%>
     # @return [::Gapic::Config::Method]
     #
     attr_reader :<%= method.name %>
diff --git a/gapic-generator/templates/default/service/client/method/_def.text.erb b/gapic-generator/templates/default/service/client/method/_def.text.erb
index 86aacfea9..5c718824a 100644
--- a/gapic-generator/templates/default/service/client/method/_def.text.erb
+++ b/gapic-generator/templates/default/service/client/method/_def.text.erb
@@ -13,11 +13,13 @@
 #
 <%= render partial: "service/client/method/docs/snippets", locals: { method: method } -%>
 <%= render partial: "service/client/method/docs/samples", locals: { method: method } -%>
-def <%= method.name %> request, options = nil
+def <%= method.name %> request<%= method.resumable_upload? ? " = {}" : "" %>, options = nil
 <%= indent render(partial: "service/client/method/def/request", locals: { method: method }), 2 %>
 
 <%= indent render(partial: "service/client/method/def/options_defaults", locals: { method: method }), 2 %>
 
 <%= indent render(partial: "service/client/method/def/response", locals: { method: method }), 2 %>
+<%- unless method.resumable_upload? -%>
 <%= render partial: "service/client/method/def/rescue", locals: { method: method } -%>
+<%- end -%>
 end
diff --git a/gapic-generator/templates/default/service/client/method/def/_options_defaults.text.erb b/gapic-generator/templates/default/service/client/method/def/_options_defaults.text.erb
index 61e227017..c745bcab8 100644
--- a/gapic-generator/templates/default/service/client/method/def/_options_defaults.text.erb
+++ b/gapic-generator/templates/default/service/client/method/def/_options_defaults.text.erb
@@ -8,7 +8,12 @@ metadata = @config.rpcs.<%= method.name %>.metadata.to_h
 # Set x-goog-api-client, x-goog-user-project and x-goog-api-version headers
 metadata[:"x-goog-api-client"] ||= ::Gapic::Headers.x_goog_api_client \
   lib_name: @config.lib_name, lib_version: @config.lib_version,
+<%- if method.resumable_upload? -%>
+  gapic_version: ::<%= method.service.gem.version_name_full %>,
+  transports_version_send: [:rest]
+<%- else -%>
   gapic_version: ::<%= method.service.gem.version_name_full %>
+<%- end -%>
 metadata[:"x-goog-api-version"] = API_VERSION unless API_VERSION.empty?
 metadata[:"x-goog-user-project"] = @quota_project_id if @quota_project_id
 <%- if method.routing_params? && !method.client_streaming? -%>
diff --git a/gapic-generator/templates/default/service/client/method/def/_response.text.erb b/gapic-generator/templates/default/service/client/method/def/_response.text.erb
index 351a7e69d..1f7868fde 100644
--- a/gapic-generator/templates/default/service/client/method/def/_response.text.erb
+++ b/gapic-generator/templates/default/service/client/method/def/_response.text.erb
@@ -1,5 +1,7 @@
 <%- assert_locals method -%>
-<%- if method.paged? -%>
+<%- if method.resumable_upload? -%>
+<%= render partial: "service/client/method/def/response_resumable_upload", locals: { method: method } -%>
+<%- elsif method.paged? -%>
 <%= render partial: "service/client/method/def/response_paged", locals: { method: method } -%>
 <%- elsif method.rest.nonstandard_lro? -%>
 <%= render partial: "service/client/method/def/response_nonstandard_lro", locals: { method: method } -%>
diff --git a/gapic-generator/templates/default/service/client/method/def/_response_resumable_upload.text.erb b/gapic-generator/templates/default/service/client/method/def/_response_resumable_upload.text.erb
new file mode 100644
index 000000000..9a223031f
--- /dev/null
+++ b/gapic-generator/templates/default/service/client/method/def/_response_resumable_upload.text.erb
@@ -0,0 +1,13 @@
+<%- assert_locals method -%>
+<%- error_handler = render(partial: "service/client/method/def/upload_error_handler", locals: { method: method }).strip -%>
+::Gapic::ResumableUpload.new(
+  client_stub_proc:     -> { <%= method.service.resumable_upload_stub_ivar %>.client_stub },
+  initial_request_proc: lambda {
+    <%= method.service.resumable_upload_stub_name_full %>.<%= method.rest.transcoding_helper_name %> request
+  },
+  initial_headers:      options.metadata,
+  start_retry_policy:   ::Gapic::Rest::ResumableUpload.start_retry_policy_for(options),
+  response_type:        <%= method.return_type %>,
+  method_name:          "<%= method.name %>",
+  error_handler:        <%= error_handler %>
+)
diff --git a/gapic-generator/templates/default/service/client/method/def/_upload_error_handler.text.erb b/gapic-generator/templates/default/service/client/method/def/_upload_error_handler.text.erb
new file mode 100644
index 000000000..9fc15293a
--- /dev/null
+++ b/gapic-generator/templates/default/service/client/method/def/_upload_error_handler.text.erb
@@ -0,0 +1,2 @@
+<%- assert_locals method -%>
+nil
diff --git a/gapic-generator/templates/default/service/client/method/docs/_request_normal.text.erb b/gapic-generator/templates/default/service/client/method/docs/_request_normal.text.erb
index 1f9c7699b..146e34a2e 100644
--- a/gapic-generator/templates/default/service/client/method/docs/_request_normal.text.erb
+++ b/gapic-generator/templates/default/service/client/method/docs/_request_normal.text.erb
@@ -1,4 +1,22 @@
 <%- assert_locals method -%>
+<%- if method.resumable_upload? -%>
+# @overload <%= method.name %>(request = {}, options = nil)
+#   Pass arguments to `<%= method.name %>` via a request object, either of type
+#   {<%= method.request_type %>} or an equivalent Hash.
+#
+#   @param request [<%= method.request_type %>, ::Hash]
+#     A request object representing the call parameters. Optional: it is ignored when the returned
+#     handle is resumed rather than started, because a resumed upload targets a session the server
+#     has already created.
+#   @param options [::Gapic::CallOptions, ::Hash]
+#     Overrides for the initiation request only. The `timeout`, `retry_policy` and `metadata` set
+#     here apply to the single request that creates the upload session, not to the upload as a
+#     whole: an upload still transferring bytes an hour later has long outlived this `timeout`.
+#     To bound the whole upload, pass `upload_timeout:` to {::Gapic::ResumableUpload#start} or
+#     {::Gapic::ResumableUpload#resume}. Chunk transfers are retried by the upload protocol itself,
+#     with the defaults documented on {::Gapic::ResumableUpload}; no call option reaches them.
+#     Optional.
+<%- else -%>
 # @overload <%= method.name %>(request, options = nil)
 #   Pass arguments to `<%= method.name %>` via a request object, either of type
 #   {<%= method.request_type %>} or an equivalent Hash.
@@ -8,6 +26,7 @@
 #     parameters, or to keep all the default parameter values, pass an empty Hash.
 #   @param options [::Gapic::CallOptions, ::Hash]
 #     Overrides the default settings for this call, e.g, timeout, retries, etc. Optional.
+<%- end -%>
 <%-if method.arguments.any?-%>
 #
 <%- arg_list = method.arguments.map { |arg| "#{arg.name}: nil"}.join ", " -%>
diff --git a/gapic-generator/templates/default/service/client/method/docs/_response.text.erb b/gapic-generator/templates/default/service/client/method/docs/_response.text.erb
index d428567e2..eed8d9d05 100644
--- a/gapic-generator/templates/default/service/client/method/docs/_response.text.erb
+++ b/gapic-generator/templates/default/service/client/method/docs/_response.text.erb
@@ -1,6 +1,12 @@
 <%- assert_locals method -%>
+<%- if method.resumable_upload? -%>
+# @return [::Gapic::ResumableUpload]
+#   A reusable upload handle. No request is sent and no byte is read from a stream until
+#   {::Gapic::ResumableUpload#start} or {::Gapic::ResumableUpload#resume} is called on it.
+<%- else -%>
 # @yield [response, operation] Access the result along with the RPC operation
 # @yieldparam response [<%= method.doc_response_type %>]
 # @yieldparam operation [::GRPC::ActiveCall::Operation]
 #
 # @return [<%= method.doc_response_type %>]
+<%- end -%>
diff --git a/gapic-generator/templates/default/service/rest/client/_client.text.erb b/gapic-generator/templates/default/service/rest/client/_client.text.erb
index c733679ae..b5dd9c6d3 100644
--- a/gapic-generator/templates/default/service/rest/client/_client.text.erb
+++ b/gapic-generator/templates/default/service/rest/client/_client.text.erb
@@ -21,6 +21,14 @@ require "securerandom"
 <%= indent service.doc_description(transport: :rest), "# " %>
 #
 <%- end -%>
+<%- if service.resumable_upload? -%>
+<%- upload_names = service.resumable_upload_methods.map { |upload_method| "`#{upload_method.name}`" }.join ", " -%>
+<%- upload_many = service.resumable_upload_methods.length > 1 -%>
+# <%= upload_names %> <%= upload_many ? "perform resumable uploads: each returns" : "performs a resumable upload: it returns" %> a {::Gapic::ResumableUpload} handle
+# instead of a response, and the per-call `timeout` and retry policy cover only the request that
+# creates the upload session, not the upload itself.
+#
+<%- end -%>
 class <%= service.rest.client_name %>
   # @private
   API_VERSION = "<%= service.api_version %>".freeze
@@ -141,6 +149,16 @@ class <%= service.rest.client_name %>
       <%- end -%>
     end
 
+    <%- end -%>
+    <%- if service.resumable_upload? -%>
+    <%= service.resumable_upload_stub_ivar %> = <%= service.resumable_upload_stub_name_full %>.new(
+      endpoint: @config.endpoint,
+      endpoint_template: DEFAULT_ENDPOINT_TEMPLATE,
+      universe_domain: @config.universe_domain,
+      credentials: credentials,
+      logger: @config.logger
+    )
+
     <%- end -%>
     @<%= service.stub_name %> = <%= service.rest.service_stub_name_full %>.new(
       endpoint: @config.endpoint,
diff --git a/gapic-generator/templates/default/service/rest/client/_config.text.erb b/gapic-generator/templates/default/service/rest/client/_config.text.erb
index 364e6d8cd..cf38dc384 100644
--- a/gapic-generator/templates/default/service/rest/client/_config.text.erb
+++ b/gapic-generator/templates/default/service/rest/client/_config.text.erb
@@ -162,6 +162,14 @@ class Configuration
   <%- method_service.rest.methods.each do |method| -%>
     ##
     # RPC-specific configuration for `<%= method.name %>`
+  <%- if method.resumable_upload? -%>
+    #
+    # `<%= method.name %>` performs a resumable upload, so the `timeout` and `retry_policy`
+    # configured here cover the request that creates the upload session only, not the upload
+    # itself. The whole-upload budget is the `upload_timeout:` argument of
+    # {::Gapic::ResumableUpload#start} and {::Gapic::ResumableUpload#resume}.
+    #
+  <%- end -%>
     # @return [::Gapic::Config::Method]
     #
     attr_reader :<%= method.name %>
diff --git a/gapic-generator/templates/default/service/rest/client/method/_def.text.erb b/gapic-generator/templates/default/service/rest/client/method/_def.text.erb
index b21bcef94..7831a0721 100644
--- a/gapic-generator/templates/default/service/rest/client/method/_def.text.erb
+++ b/gapic-generator/templates/default/service/rest/client/method/_def.text.erb
@@ -10,7 +10,7 @@
 <%= render partial: "service/rest/client/method/docs/error", locals: { method: method } -%>
 #
 <%= render partial: "service/rest/client/method/docs/snippets", locals: { method: method } -%>
-def <%= method.name %> request, options = nil
+def <%= method.name %> request<%= method.resumable_upload? ? " = {}" : "" %>, options = nil
 <%= indent render(partial: "service/client/method/def/request", locals: { method: method }), 2 %>
 
 <%= indent render(partial: "service/rest/client/method/def/options_defaults", locals: { method: method }), 2 %>
@@ -20,5 +20,7 @@ def <%= method.name %> request, options = nil
 <%- end -%>
 
 <%= indent render(partial: "service/rest/client/method/def/response", locals: { method: method }), 2 %>
+<%- unless method.resumable_upload? -%>
 <%= render partial: "service/rest/client/method/def/rescue", locals: { method: method } -%>
+<%- end -%>
 end
diff --git a/gapic-generator/templates/default/service/rest/client/method/def/_response.text.erb b/gapic-generator/templates/default/service/rest/client/method/def/_response.text.erb
index 2bb108fd6..a4e28c0e5 100644
--- a/gapic-generator/templates/default/service/rest/client/method/def/_response.text.erb
+++ b/gapic-generator/templates/default/service/rest/client/method/def/_response.text.erb
@@ -1,5 +1,7 @@
 <%- assert_locals method -%>
-<%- if method.rest.paged? -%>
+<%- if method.resumable_upload? -%>
+<%= render partial: "service/rest/client/method/def/response_resumable_upload", locals: { method: method } -%>
+<%- elsif method.rest.paged? -%>
 <%= render partial: "service/rest/client/method/def/response_paged", locals: { method: method } -%>
 <%- elsif method.rest.nonstandard_lro? -%>
 <%= render partial: "service/rest/client/method/def/response_nonstandard_lro", locals: { method: method } -%>
diff --git a/gapic-generator/templates/default/service/rest/client/method/def/_response_resumable_upload.text.erb b/gapic-generator/templates/default/service/rest/client/method/def/_response_resumable_upload.text.erb
new file mode 100644
index 000000000..62afcba12
--- /dev/null
+++ b/gapic-generator/templates/default/service/rest/client/method/def/_response_resumable_upload.text.erb
@@ -0,0 +1,13 @@
+<%- assert_locals method -%>
+<%- error_handler = render(partial: "service/rest/client/method/def/upload_error_handler", locals: { method: method }).strip -%>
+::Gapic::ResumableUpload.new(
+  client_stub_proc:     -> { <%= method.service.resumable_upload_stub_ivar %>.client_stub },
+  initial_request_proc: lambda {
+    <%= method.service.resumable_upload_stub_name_full %>.<%= method.rest.transcoding_helper_name %> request
+  },
+  initial_headers:      options.metadata,
+  start_retry_policy:   ::Gapic::Rest::ResumableUpload.start_retry_policy_for(options),
+  response_type:        <%= method.return_type %>,
+  method_name:          "<%= method.name %>",
+  error_handler:        <%= error_handler %>
+)
diff --git a/gapic-generator/templates/default/service/rest/client/method/def/_upload_error_handler.text.erb b/gapic-generator/templates/default/service/rest/client/method/def/_upload_error_handler.text.erb
new file mode 100644
index 000000000..9fc15293a
--- /dev/null
+++ b/gapic-generator/templates/default/service/rest/client/method/def/_upload_error_handler.text.erb
@@ -0,0 +1,2 @@
+<%- assert_locals method -%>
+nil
diff --git a/gapic-generator/templates/default/service/rest/client/method/docs/_request.text.erb b/gapic-generator/templates/default/service/rest/client/method/docs/_request.text.erb
index 2e170934d..e69a0340a 100644
--- a/gapic-generator/templates/default/service/rest/client/method/docs/_request.text.erb
+++ b/gapic-generator/templates/default/service/rest/client/method/docs/_request.text.erb
@@ -1,4 +1,22 @@
 <%- assert_locals method -%>
+<%- if method.resumable_upload? -%>
+# @overload <%= method.name %>(request = {}, options = nil)
+#   Pass arguments to `<%= method.name %>` via a request object, either of type
+#   {<%= method.request_type %>} or an equivalent Hash.
+#
+#   @param request [<%= method.request_type %>, ::Hash]
+#     A request object representing the call parameters. Optional: it is ignored when the returned
+#     handle is resumed rather than started, because a resumed upload targets a session the server
+#     has already created.
+#   @param options [::Gapic::CallOptions, ::Hash]
+#     Overrides for the initiation request only. The `timeout`, `retry_policy` and `metadata` set
+#     here apply to the single request that creates the upload session, not to the upload as a
+#     whole: an upload still transferring bytes an hour later has long outlived this `timeout`.
+#     To bound the whole upload, pass `upload_timeout:` to {::Gapic::ResumableUpload#start} or
+#     {::Gapic::ResumableUpload#resume}. Chunk transfers are retried by the upload protocol itself,
+#     with the defaults documented on {::Gapic::ResumableUpload}; no call option reaches them.
+#     Optional.
+<%- else -%>
 # @overload <%= method.name %>(request, options = nil)
 #   Pass arguments to `<%= method.name %>` via a request object, either of type
 #   {<%= method.request_type %>} or an equivalent Hash.
@@ -8,6 +26,7 @@
 #     parameters, or to keep all the default parameter values, pass an empty Hash.
 #   @param options [::Gapic::CallOptions, ::Hash]
 #     Overrides the default settings for this call, e.g, timeout, retries etc. Optional.
+<%- end -%>
 <%-if method.arguments.any?-%>
 #
 <%- arg_list = method.arguments.map { |arg| "#{arg.name}: nil"}.join ", " -%>
diff --git a/gapic-generator/templates/default/service/rest/client/method/docs/_result.text.erb b/gapic-generator/templates/default/service/rest/client/method/docs/_result.text.erb
index e28ce8321..6d367aa9a 100644
--- a/gapic-generator/templates/default/service/rest/client/method/docs/_result.text.erb
+++ b/gapic-generator/templates/default/service/rest/client/method/docs/_result.text.erb
@@ -1,5 +1,9 @@
 <%- assert_locals method -%>
-<%- if method.server_streaming? -%>
+<%- if method.resumable_upload? -%>
+# @return [::Gapic::ResumableUpload]
+#   A reusable upload handle. No request is sent and no byte is read from a stream until
+#   {::Gapic::ResumableUpload#start} or {::Gapic::ResumableUpload#resume} is called on it.
+<%- elsif method.server_streaming? -%>
 # @return [::Enumerable<<%= method.rest.doc_response_type %>>]
 <%- else -%>
 # @yield [result, operation] Access the result along with the TransportOperation object
diff --git a/gapic-generator/templates/default/service/rest/service_stub/_service_stub.text.erb b/gapic-generator/templates/default/service/rest/service_stub/_service_stub.text.erb
index 7fc6bd10a..4de434bd7 100644
--- a/gapic-generator/templates/default/service/rest/service_stub/_service_stub.text.erb
+++ b/gapic-generator/templates/default/service/rest/service_stub/_service_stub.text.erb
@@ -52,11 +52,11 @@ class <%= service.rest.service_stub_name %>
     stub ? @client_stub.stub_logger : @client_stub.logger
   end
 
-  <%- service.rest.methods.each do |method| -%>
+  <%- service.rest.service_stub_methods.each do |method| -%>
   <%= indent_tail render(partial: "service/rest/service_stub/method/def", locals: { method: method, service_stub_name: service.rest.service_stub_name }), 2 %>
 
   <%- end %>
-  <%- service.rest.methods.each do |method| -%>
+  <%- service.rest.service_stub_methods.each do |method| -%>
   <%= indent_tail render(partial: "service/rest/service_stub/grpc_transcoding_method/def", locals: { method: method }), 2 %>
 
   <%- end %>
diff --git a/gapic-generator/templates/default/service/rest/service_stub/grpc_transcoding_method/_bindings.text.erb b/gapic-generator/templates/default/service/rest/service_stub/grpc_transcoding_method/_bindings.text.erb
new file mode 100644
index 000000000..d210eac8c
--- /dev/null
+++ b/gapic-generator/templates/default/service/rest/service_stub/grpc_transcoding_method/_bindings.text.erb
@@ -0,0 +1,15 @@
+<%- assert_locals method -%>
+<%- method.http_bindings.each do |http_binding| -%>
+.with_bindings(
+  uri_method: :<%= http_binding.verb %>,
+  uri_template: "<%= http_binding.uri_for_transcoding %>",
+<%- if http_binding.body? -%>
+  body: "<%= http_binding.body %>",
+<%- end -%>
+  matches: [
+<%- http_binding.routing_params_transcoder_matches_strings.each do |match_str| -%>
+    <%= match_str %>
+<%- end -%>
+  ]
+)
+<%- end -%>
diff --git a/gapic-generator/templates/default/service/rest/service_stub/grpc_transcoding_method/_def.text.erb b/gapic-generator/templates/default/service/rest/service_stub/grpc_transcoding_method/_def.text.erb
index a5776f718..4a4377f7a 100644
--- a/gapic-generator/templates/default/service/rest/service_stub/grpc_transcoding_method/_def.text.erb
+++ b/gapic-generator/templates/default/service/rest/service_stub/grpc_transcoding_method/_def.text.erb
@@ -19,19 +19,6 @@ def self.<%= method.rest.transcoding_helper_name %> request_pb<%= boverr_str %>
   <%- end %>
   <%- assignment = method.service.rest.is_main_mixin_service? ? "||=" : "=" -%>
   transcoder <%= assignment %> Gapic::Rest::GrpcTranscoder.new
-  <%- method.http_bindings.each do |http_binding| -%>
-                                            .with_bindings(
-                                              uri_method: :<%= http_binding.verb %>,
-                                              uri_template: "<%= http_binding.uri_for_transcoding %>",
-                                            <%- if http_binding.body? -%>
-                                              body: "<%= http_binding.body %>",
-                                            <%- end -%>
-                                              matches: [
-                                            <%- http_binding.routing_params_transcoder_matches_strings.each do |match_str| -%>
-                                                <%= match_str %>
-                                            <%- end -%>
-                                              ]
-                                            )
-  <%- end %>
+<%= indent render(partial: "service/rest/service_stub/grpc_transcoding_method/bindings", locals: { method: method }), 44 %>
   transcoder.transcode request_pb
 end
diff --git a/gapic-generator/templates/default/service/rest/test/client.text.erb b/gapic-generator/templates/default/service/rest/test/client.text.erb
index 10d3ab560..c8bc2f48a 100644
--- a/gapic-generator/templates/default/service/rest/test/client.text.erb
+++ b/gapic-generator/templates/default/service/rest/test/client.text.erb
@@ -9,7 +9,7 @@ require "<%= service.rest.service_require %>"
 class <%= service.rest.client_name_full %>Test < Minitest::Test
 <%= indent render(partial: "service/rest/test/method/setup"), 2 %>
 
-<% service.rest.methods.each do |method| %>
+<% service.rest.methods.reject(&:resumable_upload?).each do |method| %>
 <%= indent render(partial: "service/rest/test/method/#{method.kind}",
                   locals: { method: method }), 2 %>
 
diff --git a/gapic-generator/templates/default/service/resumable_upload_stub.text.erb b/gapic-generator/templates/default/service/resumable_upload_stub.text.erb
new file mode 100644
index 000000000..c27d444fb
--- /dev/null
+++ b/gapic-generator/templates/default/service/resumable_upload_stub.text.erb
@@ -0,0 +1,6 @@
+<%- assert_locals service -%>
+<%= render partial: "service/resumable_upload_stub/resumable_upload_stub",
+           layout: "layouts/ruby",
+           locals: { service: service,
+                     namespace: service.service_name_full }
+%>
diff --git a/gapic-generator/templates/default/service/resumable_upload_stub/_resumable_upload_stub.text.erb b/gapic-generator/templates/default/service/resumable_upload_stub/_resumable_upload_stub.text.erb
new file mode 100644
index 000000000..5d95fb0e7
--- /dev/null
+++ b/gapic-generator/templates/default/service/resumable_upload_stub/_resumable_upload_stub.text.erb
@@ -0,0 +1,64 @@
+<%- assert_locals service -%>
+##
+# Resumable upload stub for the <%= service.name %> service.
+#
+# Owns the REST client stub the uploads are performed through, the transcoders that build their
+# initiation requests, and the credentials error that a client unable to perform REST calls has to
+# raise when an upload is attempted.
+#
+# @private
+#
+class <%= service.resumable_upload_stub_name %>
+<%- service.resumable_upload_methods.each do |method| -%>
+  # @private
+  <%= method.upload_url_prefix_const_name %> = "<%= method.upload_url_prefix %>"
+
+<%- end -%>
+  # @private
+  def initialize endpoint:, endpoint_template:, universe_domain:, credentials:, logger:
+    # These require statements are intentionally placed here to initialize
+    # the REST modules only when it's required.
+    require "gapic/rest"
+
+    # Uploads travel over REST, and credentials cannot be recovered from a gRPC channel. Rather than
+    # raise while a client is being constructed, which would break every other method on it, the
+    # error is stored and raised when an upload is actually attempted.
+    if defined?(::GRPC::Core::Channel) &&
+       (credentials.is_a?(::GRPC::Core::Channel) || credentials.is_a?(::GRPC::Core::ChannelCredentials))
+      @credentials_error = ::ArgumentError.new(
+        "Resumable uploads are performed over REST and cannot use a gRPC channel as credentials, " \
+        "because credentials cannot be recovered from a channel. Construct the client with Google " \
+        "OAuth credentials (a keyfile path, a Hash, or a Google::Auth::Credentials instance) to use " \
+        "resumable upload methods."
+      )
+      return
+    end
+
+    @client_stub = ::Gapic::Rest::ClientStub.new endpoint: endpoint,
+                                                 endpoint_template: endpoint_template,
+                                                 universe_domain: universe_domain,
+                                                 credentials: credentials,
+                                                 numeric_enums: false,
+                                                 service_name: self.class,
+                                                 raise_faraday_errors: false,
+                                                 logger: logger
+  end
+
+  ##
+  # @private
+  #
+  # The REST client stub the uploads are performed through.
+  #
+  # @return [::Gapic::Rest::ClientStub]
+  #
+  # @raise [::ArgumentError] if this client cannot perform REST calls.
+  #
+  def client_stub
+    raise @credentials_error if @credentials_error
+    @client_stub
+  end
+<%- service.resumable_upload_methods.each do |method| -%>
+
+  <%= indent_tail render(partial: "service/resumable_upload_stub/transcoding_method", locals: { method: method, service: service }), 2 %>
+<%- end -%>
+end
diff --git a/gapic-generator/templates/default/service/resumable_upload_stub/_transcoding_method.text.erb b/gapic-generator/templates/default/service/resumable_upload_stub/_transcoding_method.text.erb
new file mode 100644
index 000000000..8cfdb41ce
--- /dev/null
+++ b/gapic-generator/templates/default/service/resumable_upload_stub/_transcoding_method.text.erb
@@ -0,0 +1,22 @@
+<%- assert_locals method, service -%>
+##
+# @private
+#
+# GRPC transcoding helper method for the <%= method.name %> resumable upload initiation request.
+#
+# The upload itself is not transcoded: only the request that creates the upload session is, and the
+# verb is discarded because initiation is always a POST. Query string parameters are folded into the
+# URL, since the upload sends the initiation request with no separate parameters.
+#
+# @param request_pb [<%= method.request_type %>]
+#   A request object representing the call parameters. Required.
+# @return [Array(String, [String, nil])]
+#   Uri, Body
+def self.<%= method.rest.transcoding_helper_name %> request_pb
+  transcoder = ::Gapic::Rest::GrpcTranscoder.new
+<%= indent render(partial: "service/rest/service_stub/grpc_transcoding_method/bindings", locals: { method: method }), 46 %>
+  _verb, uri, query_string_params, body = transcoder.transcode request_pb
+  uri = "/#{<%= method.upload_url_prefix_const_name %>}#{uri}"
+  uri = "#{uri}?#{query_string_params.join '&'}" if query_string_params.any?
+  [uri, body]
+end
diff --git a/gapic-generator/templates/default/service/test/client.text.erb b/gapic-generator/templates/default/service/test/client.text.erb
index 6b3e218c0..50ace32ce 100644
--- a/gapic-generator/templates/default/service/test/client.text.erb
+++ b/gapic-generator/templates/default/service/test/client.text.erb
@@ -10,7 +10,7 @@ require "<%= service.service_require %>"
 class <%= service.client_name_full %>Test < Minitest::Test
 <%= indent render(partial: "service/test/method/setup"), 2 %>
 
-<% service.methods.each do |method| %>
+<% service.methods.reject(&:resumable_upload?).each do |method| %>
 <%= indent render(partial: "service/test/method/#{method.kind}",
                   locals: { method: method }), 2 %>
 
diff --git a/gapic-generator/templates/default/service/test/resumable_upload.text.erb b/gapic-generator/templates/default/service/test/resumable_upload.text.erb
new file mode 100644
index 000000000..7d421f6c1
--- /dev/null
+++ b/gapic-generator/templates/default/service/test/resumable_upload.text.erb
@@ -0,0 +1,132 @@
+<%- assert_locals service -%>
+<%= render partial: "shared/header" %>
+require "helper"
+
+require "gapic/rest"
+<%- if service.generate_grpc_clients? -%>
+require "gapic/grpc/service_stub"
+<%- end -%>
+
+require "<%= service.proto_service_require %>"
+<%- if service.generate_grpc_clients? -%>
+require "<%= service.service_require %>"
+<%- end -%>
+<%- if service.generate_rest_clients? -%>
+require "<%= service.rest.service_require %>"
+<%- end -%>
+
+class <%= service.service_name_full %>::ResumableUploadTest < Minitest::Test
+  ##
+  # Stands in for the transport stub a client builds in its constructor. An upload method sends
+  # nothing, so this only has to satisfy the constructor.
+  #
+  class ClientStub
+    def endpoint
+      "endpoint.example.com"
+    end
+
+    def universe_domain
+      "example.com"
+    end
+
+    def stub_logger
+      nil
+    end
+
+    def logger
+      nil
+    end
+  end
+<%- service.resumable_upload_methods.each do |method| -%>
+<%- fields = method.fields_with_first_oneof -%>
+<%- prefix_const = "#{service.resumable_upload_stub_name_full}::#{method.upload_url_prefix_const_name}" -%>
+
+  def test_transcode_<%= method.name %>_request
+    request = <%= method.request_type %>.new
+
+    # The transcoder is stubbed so the test pins the upload-specific post-processing (prefix,
+    # query-string folding, [uri, body] shape) independently of the method's path template.
+    transcoder = Gapic::Rest::GrpcTranscoder.new
+    transcoder.stub :transcode, [:post, "/transcoded/path", ["foo=bar", "baz=qux"], "{\"body\":true}"] do
+      Gapic::Rest::GrpcTranscoder.stub :new, transcoder do
+        uri, body = <%= service.resumable_upload_stub_name_full %>.<%= method.rest.transcoding_helper_name %> request
+
+        assert_equal "/#{<%= prefix_const %>}/transcoded/path?foo=bar&baz=qux", uri
+        assert_equal "{\"body\":true}", body
+      end
+    end
+  end
+<%- if service.generate_grpc_clients? -%>
+
+  def test_<%= method.name %>
+    <%- fields.each do |field| -%>
+    <%= field.value_name %> = <%= field.default_value %>
+    <%- end -%>
+
+    client = nil
+    Gapic::ServiceStub.stub :new, ClientStub.new do
+      client = <%= service.client_name_full %>.new do |config|
+        config.credentials = :this_channel_is_insecure
+      end
+    end
+
+    upload = client.<%= method.name %> <%= method.request_type %>.new(<%= fields.map(&:as_kwarg).join ", " %>)
+
+    # The call builds a handle and sends nothing. The upload only begins on #start or #resume.
+    assert_kind_of ::Gapic::ResumableUpload, upload
+    assert_nil upload.resume_handle
+    refute upload.resumable?
+    refute upload.running?
+  end
+
+  def test_<%= method.name %>_channel_credentials
+    <%- fields.each do |field| -%>
+    <%= field.value_name %> = <%= field.default_value %>
+    <%- end -%>
+    grpc_channel = GRPC::Core::Channel.new "localhost:8888", nil, :this_channel_is_insecure
+
+    client = nil
+    Gapic::ServiceStub.stub :new, ClientStub.new do
+      client = <%= service.client_name_full %>.new do |config|
+        config.credentials = grpc_channel
+      end
+    end
+
+    upload = client.<%= method.name %> <%= method.request_type %>.new(<%= fields.map(&:as_kwarg).join ", " %>)
+
+    assert_kind_of ::Gapic::ResumableUpload, upload
+
+    # Uploads travel over REST and cannot use a channel as credentials. The error surfaces here,
+    # rather than at construction, and before the stream is touched.
+    stream = StringIO.new "resumable upload test payload"
+    assert_raises ::ArgumentError do
+      upload.start stream: stream, content_type: "application/octet-stream"
+    end
+    assert_equal 0, stream.pos
+  end
+<%- end -%>
+<%- if service.generate_rest_clients? -%>
+
+  def test_<%= method.name %>_rest
+    <%- fields.each do |field| -%>
+    <%= field.value_name %> = <%= field.default_value %>
+    <%- end -%>
+
+    client = nil
+    Gapic::Rest::ClientStub.stub :new, ClientStub.new do
+      client = <%= service.rest.client_name_full %>.new do |config|
+        config.credentials = :dummy_value
+      end
+    end
+
+    upload = client.<%= method.name %> <%= method.request_type %>.new(<%= fields.map(&:as_kwarg).join ", " %>)
+
+    # The call builds a handle and sends nothing. The upload only begins on #start or #resume.
+    assert_kind_of ::Gapic::ResumableUpload, upload
+    assert_nil upload.resume_handle
+    refute upload.resumable?
+    refute upload.running?
+  end
+<%- end -%>
+<%- end -%>
+end
diff --git a/gapic-generator/test/gapic/presenters/method/resumable_upload_test.rb b/gapic-generator/test/gapic/presenters/method/resumable_upload_test.rb
new file mode 100644
index 000000000..1498d50a3
--- /dev/null
+++ b/gapic-generator/test/gapic/presenters/method/resumable_upload_test.rb
@@ -0,0 +1,191 @@
+# frozen_string_literal: true
+
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+require "test_helper"
+
+class MethodPresenterResumableUploadTest < PresenterTest
+  def test_showcase_UploadMedia
+    presenter = method_presenter :showcase, "ResumableUploadService", "UploadMedia"
+
+    assert presenter.resumable_upload?
+    assert_equal "resumable/upload", presenter.upload_url_prefix
+  end
+
+  def test_showcase_Echo_is_not_an_upload
+    presenter = method_presenter :showcase, "Echo", "Echo"
+
+    refute presenter.resumable_upload?
+    assert_nil presenter.upload_url_prefix
+  end
+
+  def test_googleads_CreateYouTubeVideoUpload
+    presenter = method_presenter :googleads, "YouTubeVideoUploadService", "CreateYouTubeVideoUpload"
+
+    assert presenter.resumable_upload?
+    assert_equal "resumable/upload", presenter.upload_url_prefix
+  end
+
+  def test_googleads_siblings_are_not_uploads
+    %w[UpdateYouTubeVideoUpload RemoveYouTubeVideoUpload].each do |method_name|
+      presenter = method_presenter :googleads, "YouTubeVideoUploadService", method_name
+
+      refute presenter.resumable_upload?, "#{method_name} must not be detected as an upload"
+      assert_nil presenter.upload_url_prefix
+    end
+  end
+end
+
+class ResumableUploadModelPrefixTest < Minitest::Test
+  Model = Gapic::Model::Method::ResumableUpload
+
+  def test_exact_match
+    assert_equal "resumable/upload",
+                 Model.url_prefix_for("google.showcase.v1beta1.ResumableUploadService.UploadMedia")
+  end
+
+  def test_versioned_match_covers_every_ads_version
+    %w[v23 v24 v25 v23_1 v101].each do |version|
+      full_name = "google.ads.googleads.#{version}.services.YouTubeVideoUploadService.CreateYouTubeVideoUpload"
+
+      assert_equal "resumable/upload", Model.url_prefix_for(full_name), "#{version} must match"
+    end
+  end
+
+  def test_versioned_match_does_not_constrain_the_middle_segments
+    assert_equal "resumable/upload",
+                 Model.url_prefix_for("google.ads.googleads.v25.YouTubeVideoUploadService.CreateYouTubeVideoUpload")
+  end
+
+  def test_near_misses_do_not_match
+    [
+      # Wrong package.
+      "google.ads.googleadsx.v25.services.YouTubeVideoUploadService.CreateYouTubeVideoUpload",
+      # Unversioned package.
+      "google.ads.googleads.services.YouTubeVideoUploadService.CreateYouTubeVideoUpload",
+      # Right service, wrong method.
+      "google.ads.googleads.v25.services.YouTubeVideoUploadService.UpdateYouTubeVideoUpload",
+      # Right method, wrong service.
+      "google.ads.googleads.v25.services.CampaignService.CreateYouTubeVideoUpload",
+      # Right suffix, but not left-anchored at the ads package.
+      "example.google.ads.googleads.v25.services.YouTubeVideoUploadService.CreateYouTubeVideoUpload",
+      # Right showcase service, wrong method.
+      "google.showcase.v1beta1.ResumableUploadService.UploadMediaAgain"
+    ].each do |full_name|
+      assert_nil Model.url_prefix_for(full_name), "#{full_name} must not match"
+    end
+  end
+end
+
+class ResumableUploadModelValidationTest < Minitest::Test
+  Model = Gapic::Model::Method::ResumableUpload
+
+  # A stand-in for a MethodPresenter, carrying only what validation reads.
+  class FakeMethod
+    def initialize **overrides
+      @attrs = {
+        grpc_full_name:   "google.showcase.v1beta1.ResumableUploadService.UploadMedia",
+        client_streaming: false,
+        server_streaming: false,
+        paged:            false,
+        lro:              false,
+        nonstandard_lro:  false,
+        http_bindings:    [FakeBinding.new(verb: :post, body: "*")]
+      }.merge overrides
+    end
+
+    def grpc_full_name
+      @attrs[:grpc_full_name]
+    end
+
+    def client_streaming?
+      @attrs[:client_streaming]
+    end
+
+    def server_streaming?
+      @attrs[:server_streaming]
+    end
+
+    def paged?
+      @attrs[:paged]
+    end
+
+    def lro?
+      @attrs[:lro]
+    end
+
+    def nonstandard_lro?
+      @attrs[:nonstandard_lro]
+    end
+
+    def http_bindings
+      @attrs[:http_bindings]
+    end
+  end
+
+  FakeBinding = Struct.new :verb, :body, keyword_init: true do
+    def body?
+      !body.nil? && !body.empty?
+    end
+  end
+
+  def test_a_matched_unary_post_with_a_body_is_accepted
+    model = Model.create FakeMethod.new
+
+    assert_equal "resumable/upload", model.url_prefix
+  end
+
+  def test_an_unmatched_method_is_not_validated_at_all
+    assert_nil Model.create(FakeMethod.new(grpc_full_name: "google.showcase.v1beta1.Echo.Echo",
+                                           server_streaming: true))
+  end
+
+  def test_streaming_is_rejected
+    assert_rejected FakeMethod.new(client_streaming: true), "non-streaming"
+    assert_rejected FakeMethod.new(server_streaming: true), "non-streaming"
+  end
+
+  def test_pagination_is_rejected
+    assert_rejected FakeMethod.new(paged: true), "non-paginated"
+  end
+
+  def test_long_running_is_rejected
+    assert_rejected FakeMethod.new(lro: true), "long-running"
+    assert_rejected FakeMethod.new(nonstandard_lro: true), "long-running"
+  end
+
+  def test_a_missing_binding_is_rejected
+    assert_rejected FakeMethod.new(http_bindings: []), "an HTTP binding"
+  end
+
+  def test_a_non_post_binding_is_rejected
+    assert_rejected FakeMethod.new(http_bindings: [FakeBinding.new(verb: :get, body: "*")]), "POST"
+  end
+
+  def test_a_bodiless_binding_is_rejected
+    assert_rejected FakeMethod.new(http_bindings: [FakeBinding.new(verb: :post, body: nil)]), "a body"
+  end
+
+  private
+
+  def assert_rejected method, expected_reason
+    error = assert_raises Gapic::Model::ModelError do
+      Model.create method
+    end
+
+    assert_includes error.message, method.grpc_full_name
+    assert_includes error.message, expected_reason
+  end
+end
diff --git a/gapic-generator/test/gapic/presenters/service/resumable_upload_test.rb b/gapic-generator/test/gapic/presenters/service/resumable_upload_test.rb
new file mode 100644
index 000000000..9e0296e6f
--- /dev/null
+++ b/gapic-generator/test/gapic/presenters/service/resumable_upload_test.rb
@@ -0,0 +1,64 @@
+# frozen_string_literal: true
+
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+require "test_helper"
+
+class ServicePresenterResumableUploadTest < PresenterTest
+  def test_showcase_ResumableUploadService
+    presenter = service_presenter :showcase, "ResumableUploadService"
+
+    assert presenter.resumable_upload?
+    assert_equal ["upload_media"], presenter.resumable_upload_methods.map(&:name)
+
+    assert_equal "ResumableUploadStub", presenter.resumable_upload_stub_name
+    assert_equal "::Google::Showcase::V1beta1::ResumableUploadService::ResumableUploadStub",
+                 presenter.resumable_upload_stub_name_full
+    assert_equal "google/showcase/v1beta1/resumable_upload_service/resumable_upload_stub",
+                 presenter.resumable_upload_stub_require
+    assert_equal "google/showcase/v1beta1/resumable_upload_service/resumable_upload_stub.rb",
+                 presenter.resumable_upload_stub_file_path
+    assert_equal "resumable_upload_stub.rb", presenter.resumable_upload_stub_file_name
+    assert_equal "@resumable_upload_stub", presenter.resumable_upload_stub_ivar
+  end
+
+  def test_showcase_Echo_has_no_uploads
+    presenter = service_presenter :showcase, "Echo"
+
+    refute presenter.resumable_upload?
+    assert_empty presenter.resumable_upload_methods
+  end
+
+  def test_upload_rpcs_are_kept_out_of_the_rest_service_stub
+    presenter = service_presenter :showcase, "ResumableUploadService"
+
+    # The REST client still generates a method for the RPC; only the stub skips it.
+    assert_includes presenter.rest.methods.map(&:name), "upload_media"
+    refute_includes presenter.rest.service_stub_methods.map(&:name), "upload_media"
+  end
+
+  def test_non_upload_rpcs_stay_in_both_lists
+    presenter = service_presenter :showcase, "Echo"
+
+    assert_equal presenter.rest.methods.map(&:name), presenter.rest.service_stub_methods.map(&:name)
+  end
+
+  def test_googleads_YouTubeVideoUploadService
+    presenter = service_presenter :googleads, "YouTubeVideoUploadService"
+
+    assert presenter.resumable_upload?
+    assert_equal ["create_you_tube_video_upload"], presenter.resumable_upload_methods.map(&:name)
+  end
+end
diff --git a/shared/output/ads/googleads/lib/google/ads/google_ads/v25/services/you_tube_video_upload_service.rb b/shared/output/ads/googleads/lib/google/ads/google_ads/v25/services/you_tube_video_upload_service.rb
index af4dfd836..1c4048d74 100644
--- a/shared/output/ads/googleads/lib/google/ads/google_ads/v25/services/you_tube_video_upload_service.rb
+++ b/shared/output/ads/googleads/lib/google/ads/google_ads/v25/services/you_tube_video_upload_service.rb
@@ -24,6 +24,7 @@
 
 require "google/ads/google_ads/v25/services/you_tube_video_upload_service/credentials"
 require "google/ads/google_ads/v25/services/you_tube_video_upload_service/paths"
+require "google/ads/google_ads/v25/services/you_tube_video_upload_service/resumable_upload_stub"
 require "google/ads/google_ads/v25/services/you_tube_video_upload_service/client"
 
 module Google
diff --git a/shared/output/ads/googleads/lib/google/ads/google_ads/v25/services/you_tube_video_upload_service/client.rb b/shared/output/ads/googleads/lib/google/ads/google_ads/v25/services/you_tube_video_upload_service/client.rb
index 799202de9..e5e32e75d 100644
--- a/shared/output/ads/googleads/lib/google/ads/google_ads/v25/services/you_tube_video_upload_service/client.rb
+++ b/shared/output/ads/googleads/lib/google/ads/google_ads/v25/services/you_tube_video_upload_service/client.rb
@@ -30,6 +30,10 @@ module YouTubeVideoUploadService
             #
             # Service to manage YouTube video uploads.
             #
+            # `create_you_tube_video_upload` performs a resumable upload: it returns a {::Gapic::ResumableUpload} handle
+            # instead of a response, and the per-call `timeout` and retry policy cover only the request that
+            # creates the upload session, not the upload itself.
+            #
             class Client
               # @private
               API_VERSION = ""
@@ -143,6 +147,14 @@ def initialize
                 @quota_project_id = @config.quota_project
                 @quota_project_id ||= credentials.quota_project_id if credentials.respond_to? :quota_project_id
 
+                @resumable_upload_stub = ::Google::Ads::GoogleAds::V25::Services::YouTubeVideoUploadService::ResumableUploadStub.new(
+                  endpoint: @config.endpoint,
+                  endpoint_template: DEFAULT_ENDPOINT_TEMPLATE,
+                  universe_domain: @config.universe_domain,
+                  credentials: credentials,
+                  logger: @config.logger
+                )
+
                 @you_tube_video_upload_service_stub = ::Gapic::ServiceStub.new(
                   ::Google::Ads::GoogleAds::V25::Services::YouTubeVideoUploadService::Stub,
                   credentials: credentials,
@@ -181,15 +193,22 @@ def logger
               # Uploads a video to Google-managed or advertiser owned (brand) YouTube
               # channel.
               #
-              # @overload create_you_tube_video_upload(request, options = nil)
+              # @overload create_you_tube_video_upload(request = {}, options = nil)
               #   Pass arguments to `create_you_tube_video_upload` via a request object, either of type
               #   {::Google::Ads::GoogleAds::V25::Services::CreateYouTubeVideoUploadRequest} or an equivalent Hash.
               #
               #   @param request [::Google::Ads::GoogleAds::V25::Services::CreateYouTubeVideoUploadRequest, ::Hash]
-              #     A request object representing the call parameters. Required. To specify no
-              #     parameters, or to keep all the default parameter values, pass an empty Hash.
+              #     A request object representing the call parameters. Optional: it is ignored when the returned
+              #     handle is resumed rather than started, because a resumed upload targets a session the server
+              #     has already created.
               #   @param options [::Gapic::CallOptions, ::Hash]
-              #     Overrides the default settings for this call, e.g, timeout, retries, etc. Optional.
+              #     Overrides for the initiation request only. The `timeout`, `retry_policy` and `metadata` set
+              #     here apply to the single request that creates the upload session, not to the upload as a
+              #     whole: an upload still transferring bytes an hour later has long outlived this `timeout`.
+              #     To bound the whole upload, pass `upload_timeout:` to {::Gapic::ResumableUpload#start} or
+              #     {::Gapic::ResumableUpload#resume}. Chunk transfers are retried by the upload protocol itself,
+              #     with the defaults documented on {::Gapic::ResumableUpload}; no call option reaches them.
+              #     Optional.
               #
               # @overload create_you_tube_video_upload(customer_id: nil, you_tube_video_upload: nil)
               #   Pass arguments to `create_you_tube_video_upload` via keyword arguments. Note that at
@@ -201,11 +220,9 @@ def logger
               #   @param you_tube_video_upload [::Google::Ads::GoogleAds::V25::Resources::YouTubeVideoUpload, ::Hash]
               #     Required. The initial details of the video to upload. Required.
               #
-              # @yield [response, operation] Access the result along with the RPC operation
-              # @yieldparam response [::Google::Ads::GoogleAds::V25::Services::CreateYouTubeVideoUploadResponse]
-              # @yieldparam operation [::GRPC::ActiveCall::Operation]
-              #
-              # @return [::Google::Ads::GoogleAds::V25::Services::CreateYouTubeVideoUploadResponse]
+              # @return [::Gapic::ResumableUpload]
+              #   A reusable upload handle. No request is sent and no byte is read from a stream until
+              #   {::Gapic::ResumableUpload#start} or {::Gapic::ResumableUpload#resume} is called on it.
               #
               # @raise [Google::Ads::GoogleAds::Error] if the RPC is aborted.
               #
@@ -219,12 +236,18 @@ def logger
               #   request = Google::Ads::GoogleAds::V25::Services::CreateYouTubeVideoUploadRequest.new
               #
               #   # Call the create_you_tube_video_upload method.
-              #   result = client.create_you_tube_video_upload request
+              #   upload = client.create_you_tube_video_upload request
+              #
+              #   # The returned object is a handle for a resumable upload. Nothing has been
+              #   # uploaded yet, and the timeout and retry policy of the call above cover only
+              #   # the request that creates the upload session, not the upload as a whole.
+              #   stream = File.open "input.bin", "rb"
+              #   result = upload.start stream: stream, content_type: "application/octet-stream"
               #
               #   # The returned object is of type Google::Ads::GoogleAds::V25::Services::CreateYouTubeVideoUploadResponse.
               #   p result
               #
-              def create_you_tube_video_upload request, options = nil
+              def create_you_tube_video_upload request = {}, options = nil
                 raise ::ArgumentError, "request must be provided" if request.nil?
 
                 request = ::Gapic::Protobuf.coerce request, to: ::Google::Ads::GoogleAds::V25::Services::CreateYouTubeVideoUploadRequest
@@ -238,7 +261,8 @@ def create_you_tube_video_upload request, options = nil
                 # Set x-goog-api-client, x-goog-user-project and x-goog-api-version headers
                 metadata[:"x-goog-api-client"] ||= ::Gapic::Headers.x_goog_api_client \
                   lib_name: @config.lib_name, lib_version: @config.lib_version,
-                  gapic_version: ::Google::Ads::GoogleAds::VERSION
+                  gapic_version: ::Google::Ads::GoogleAds::VERSION,
+                  transports_version_send: [:rest]
                 metadata[:"x-goog-api-version"] = API_VERSION unless API_VERSION.empty?
                 metadata[:"x-goog-user-project"] = @quota_project_id if @quota_project_id
 
@@ -258,12 +282,17 @@ def create_you_tube_video_upload request, options = nil
                                        metadata:     @config.metadata,
                                        retry_policy: @config.retry_policy
 
-                @you_tube_video_upload_service_stub.call_rpc :create_you_tube_video_upload, request,
-                                                             options: options do |response, operation|
-                  yield response, operation if block_given?
-                end
-                # rescue GRPC::BadStatus => grpc_error
-                #  raise Google::Ads::GoogleAds::Error.new grpc_error.message
+                ::Gapic::ResumableUpload.new(
+                  client_stub_proc:     -> { @resumable_upload_stub.client_stub },
+                  initial_request_proc: lambda {
+                    ::Google::Ads::GoogleAds::V25::Services::YouTubeVideoUploadService::ResumableUploadStub.transcode_create_you_tube_video_upload_request request
+                  },
+                  initial_headers:      options.metadata,
+                  start_retry_policy:   ::Gapic::Rest::ResumableUpload.start_retry_policy_for(options),
+                  response_type:        ::Google::Ads::GoogleAds::V25::Services::CreateYouTubeVideoUploadResponse,
+                  method_name:          "create_you_tube_video_upload",
+                  error_handler:        nil
+                )
               end
 
               ##
@@ -637,6 +666,12 @@ def channel_pool
                 class Rpcs
                   ##
                   # RPC-specific configuration for `create_you_tube_video_upload`
+                  #
+                  # `create_you_tube_video_upload` performs a resumable upload, so the `timeout` and `retry_policy`
+                  # configured here cover the request that creates the upload session only, not the upload
+                  # itself. The whole-upload budget is the `upload_timeout:` argument of
+                  # {::Gapic::ResumableUpload#start} and {::Gapic::ResumableUpload#resume}.
+                  #
                   # @return [::Gapic::Config::Method]
                   #
                   attr_reader :create_you_tube_video_upload
diff --git a/shared/output/ads/googleads/lib/google/ads/google_ads/v25/services/you_tube_video_upload_service/resumable_upload_stub.rb b/shared/output/ads/googleads/lib/google/ads/google_ads/v25/services/you_tube_video_upload_service/resumable_upload_stub.rb
new file mode 100644
index 000000000..6a22a0737
--- /dev/null
+++ b/shared/output/ads/googleads/lib/google/ads/google_ads/v25/services/you_tube_video_upload_service/resumable_upload_stub.rb
@@ -0,0 +1,117 @@
+# frozen_string_literal: true
+
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# Auto-generated by gapic-generator-ruby. DO NOT EDIT!
+
+
+module Google
+  module Ads
+    module GoogleAds
+      module V25
+        module Services
+          module YouTubeVideoUploadService
+            ##
+            # Resumable upload stub for the YouTubeVideoUploadService service.
+            #
+            # Owns the REST client stub the uploads are performed through, the transcoders that build their
+            # initiation requests, and the credentials error that a client unable to perform REST calls has to
+            # raise when an upload is attempted.
+            #
+            # @private
+            #
+            class ResumableUploadStub
+              # @private
+              CREATE_YOU_TUBE_VIDEO_UPLOAD_URL_PREFIX = "resumable/upload"
+
+              # @private
+              def initialize endpoint:, endpoint_template:, universe_domain:, credentials:, logger:
+                # These require statements are intentionally placed here to initialize
+                # the REST modules only when it's required.
+                require "gapic/rest"
+
+                # Uploads travel over REST, and credentials cannot be recovered from a gRPC channel. Rather than
+                # raise while a client is being constructed, which would break every other method on it, the
+                # error is stored and raised when an upload is actually attempted.
+                if defined?(::GRPC::Core::Channel) &&
+                   (credentials.is_a?(::GRPC::Core::Channel) || credentials.is_a?(::GRPC::Core::ChannelCredentials))
+                  @credentials_error = ::ArgumentError.new(
+                    "Resumable uploads are performed over REST and cannot use a gRPC channel as credentials, " \
+                    "because credentials cannot be recovered from a channel. Construct the client with Google " \
+                    "OAuth credentials (a keyfile path, a Hash, or a Google::Auth::Credentials instance) to use " \
+                    "resumable upload methods."
+                  )
+                  return
+                end
+
+                @client_stub = ::Gapic::Rest::ClientStub.new endpoint: endpoint,
+                                                             endpoint_template: endpoint_template,
+                                                             universe_domain: universe_domain,
+                                                             credentials: credentials,
+                                                             numeric_enums: false,
+                                                             service_name: self.class,
+                                                             raise_faraday_errors: false,
+                                                             logger: logger
+              end
+
+              ##
+              # @private
+              #
+              # The REST client stub the uploads are performed through.
+              #
+              # @return [::Gapic::Rest::ClientStub]
+              #
+              # @raise [::ArgumentError] if this client cannot perform REST calls.
+              #
+              def client_stub
+                raise @credentials_error if @credentials_error
+                @client_stub
+              end
+
+              ##
+              # @private
+              #
+              # GRPC transcoding helper method for the create_you_tube_video_upload resumable upload initiation request.
+              #
+              # The upload itself is not transcoded: only the request that creates the upload session is, and the
+              # verb is discarded because initiation is always a POST. Query string parameters are folded into the
+              # URL, since the upload sends the initiation request with no separate parameters.
+              #
+              # @param request_pb [::Google::Ads::GoogleAds::V25::Services::CreateYouTubeVideoUploadRequest]
+              #   A request object representing the call parameters. Required.
+              # @return [Array(String, [String, nil])]
+              #   Uri, Body
+              def self.transcode_create_you_tube_video_upload_request request_pb
+                transcoder = ::Gapic::Rest::GrpcTranscoder.new
+                                                          .with_bindings(
+                                                            uri_method: :post,
+                                                            uri_template: "/v25/customers/{customer_id}/youTubeVideoUploads:create",
+                                                            body: "*",
+                                                            matches: [
+                                                              ["customer_id", %r{^[^/]+/?$}, false]
+                                                            ]
+                                                          )
+                _verb, uri, query_string_params, body = transcoder.transcode request_pb
+                uri = "/#{CREATE_YOU_TUBE_VIDEO_UPLOAD_URL_PREFIX}#{uri}"
+                uri = "#{uri}?#{query_string_params.join '&'}" if query_string_params.any?
+                [uri, body]
+              end
+            end
+          end
+        end
+      end
+    end
+  end
+end
diff --git a/shared/output/gapic/templates/showcase/lib/google/showcase/v1beta1/resumable_upload_service.rb b/shared/output/gapic/templates/showcase/lib/google/showcase/v1beta1/resumable_upload_service.rb
index 9ead063d1..c901e6c68 100644
--- a/shared/output/gapic/templates/showcase/lib/google/showcase/v1beta1/resumable_upload_service.rb
+++ b/shared/output/gapic/templates/showcase/lib/google/showcase/v1beta1/resumable_upload_service.rb
@@ -31,6 +31,7 @@
 require "google/showcase/version"
 
 require "google/showcase/v1beta1/resumable_upload_service/credentials"
+require "google/showcase/v1beta1/resumable_upload_service/resumable_upload_stub"
 require "google/showcase/v1beta1/resumable_upload_service/client"
 require "google/showcase/v1beta1/resumable_upload_service/rest"
 
diff --git a/shared/output/gapic/templates/showcase/lib/google/showcase/v1beta1/resumable_upload_service/client.rb b/shared/output/gapic/templates/showcase/lib/google/showcase/v1beta1/resumable_upload_service/client.rb
index bed24a972..0a9aba5c3 100644
--- a/shared/output/gapic/templates/showcase/lib/google/showcase/v1beta1/resumable_upload_service/client.rb
+++ b/shared/output/gapic/templates/showcase/lib/google/showcase/v1beta1/resumable_upload_service/client.rb
@@ -37,6 +37,10 @@ module ResumableUploadService
         #
         # A service showcasing universal resumable upload protocol support.
         #
+        # `upload_media` performs a resumable upload: it returns a {::Gapic::ResumableUpload} handle
+        # instead of a response, and the per-call `timeout` and retry policy cover only the request that
+        # creates the upload session, not the upload itself.
+        #
         class Client
           # @private
           API_VERSION = ""
@@ -148,6 +152,14 @@ def initialize
             @quota_project_id = @config.quota_project
             @quota_project_id ||= credentials.quota_project_id if credentials.respond_to? :quota_project_id
 
+            @resumable_upload_stub = ::Google::Showcase::V1beta1::ResumableUploadService::ResumableUploadStub.new(
+              endpoint: @config.endpoint,
+              endpoint_template: DEFAULT_ENDPOINT_TEMPLATE,
+              universe_domain: @config.universe_domain,
+              credentials: credentials,
+              logger: @config.logger
+            )
+
             @resumable_upload_service_stub = ::Gapic::ServiceStub.new(
               ::Google::Showcase::V1beta1::ResumableUploadService::Stub,
               credentials: credentials,
@@ -215,15 +227,22 @@ def logger
           ##
           # A method with media_upload annotation enabled.
           #
-          # @overload upload_media(request, options = nil)
+          # @overload upload_media(request = {}, options = nil)
           #   Pass arguments to `upload_media` via a request object, either of type
           #   {::Google::Showcase::V1beta1::UploadMediaRequest} or an equivalent Hash.
           #
           #   @param request [::Google::Showcase::V1beta1::UploadMediaRequest, ::Hash]
-          #     A request object representing the call parameters. Required. To specify no
-          #     parameters, or to keep all the default parameter values, pass an empty Hash.
+          #     A request object representing the call parameters. Optional: it is ignored when the returned
+          #     handle is resumed rather than started, because a resumed upload targets a session the server
+          #     has already created.
           #   @param options [::Gapic::CallOptions, ::Hash]
-          #     Overrides the default settings for this call, e.g, timeout, retries, etc. Optional.
+          #     Overrides for the initiation request only. The `timeout`, `retry_policy` and `metadata` set
+          #     here apply to the single request that creates the upload session, not to the upload as a
+          #     whole: an upload still transferring bytes an hour later has long outlived this `timeout`.
+          #     To bound the whole upload, pass `upload_timeout:` to {::Gapic::ResumableUpload#start} or
+          #     {::Gapic::ResumableUpload#resume}. Chunk transfers are retried by the upload protocol itself,
+          #     with the defaults documented on {::Gapic::ResumableUpload}; no call option reaches them.
+          #     Optional.
           #
           # @overload upload_media(name: nil)
           #   Pass arguments to `upload_media` via keyword arguments. Note that at
@@ -232,11 +251,9 @@ def logger
           #
           #   @param name [::String]
           #
-          # @yield [response, operation] Access the result along with the RPC operation
-          # @yieldparam response [::Google::Showcase::V1beta1::UploadMediaResponse]
-          # @yieldparam operation [::GRPC::ActiveCall::Operation]
-          #
-          # @return [::Google::Showcase::V1beta1::UploadMediaResponse]
+          # @return [::Gapic::ResumableUpload]
+          #   A reusable upload handle. No request is sent and no byte is read from a stream until
+          #   {::Gapic::ResumableUpload#start} or {::Gapic::ResumableUpload#resume} is called on it.
           #
           # @raise [::GRPC::BadStatus] if the RPC is aborted.
           #
@@ -250,12 +267,18 @@ def logger
           #   request = Google::Showcase::V1beta1::UploadMediaRequest.new
           #
           #   # Call the upload_media method.
-          #   result = client.upload_media request
+          #   upload = client.upload_media request
+          #
+          #   # The returned object is a handle for a resumable upload. Nothing has been
+          #   # uploaded yet, and the timeout and retry policy of the call above cover only
+          #   # the request that creates the upload session, not the upload as a whole.
+          #   stream = File.open "input.bin", "rb"
+          #   result = upload.start stream: stream, content_type: "application/octet-stream"
           #
           #   # The returned object is of type Google::Showcase::V1beta1::UploadMediaResponse.
           #   p result
           #
-          def upload_media request, options = nil
+          def upload_media request = {}, options = nil
             raise ::ArgumentError, "request must be provided" if request.nil?
 
             request = ::Gapic::Protobuf.coerce request, to: ::Google::Showcase::V1beta1::UploadMediaRequest
@@ -269,7 +292,8 @@ def upload_media request, options = nil
             # Set x-goog-api-client, x-goog-user-project and x-goog-api-version headers
             metadata[:"x-goog-api-client"] ||= ::Gapic::Headers.x_goog_api_client \
               lib_name: @config.lib_name, lib_version: @config.lib_version,
-              gapic_version: ::Google::Showcase::VERSION
+              gapic_version: ::Google::Showcase::VERSION,
+              transports_version_send: [:rest]
             metadata[:"x-goog-api-version"] = API_VERSION unless API_VERSION.empty?
             metadata[:"x-goog-user-project"] = @quota_project_id if @quota_project_id
 
@@ -281,9 +305,17 @@ def upload_media request, options = nil
                                    metadata:     @config.metadata,
                                    retry_policy: @config.retry_policy
 
-            @resumable_upload_service_stub.call_rpc :upload_media, request, options: options do |response, operation|
-              yield response, operation if block_given?
-            end
+            ::Gapic::ResumableUpload.new(
+              client_stub_proc:     -> { @resumable_upload_stub.client_stub },
+              initial_request_proc: lambda {
+                ::Google::Showcase::V1beta1::ResumableUploadService::ResumableUploadStub.transcode_upload_media_request request
+              },
+              initial_headers:      options.metadata,
+              start_retry_policy:   ::Gapic::Rest::ResumableUpload.start_retry_policy_for(options),
+              response_type:        ::Google::Showcase::V1beta1::UploadMediaResponse,
+              method_name:          "upload_media",
+              error_handler:        nil
+            )
           end
 
           ##
@@ -474,6 +506,12 @@ def channel_pool
             class Rpcs
               ##
               # RPC-specific configuration for `upload_media`
+              #
+              # `upload_media` performs a resumable upload, so the `timeout` and `retry_policy`
+              # configured here cover the request that creates the upload session only, not the upload
+              # itself. The whole-upload budget is the `upload_timeout:` argument of
+              # {::Gapic::ResumableUpload#start} and {::Gapic::ResumableUpload#resume}.
+              #
               # @return [::Gapic::Config::Method]
               #
               attr_reader :upload_media
diff --git a/shared/output/gapic/templates/showcase/lib/google/showcase/v1beta1/resumable_upload_service/rest.rb b/shared/output/gapic/templates/showcase/lib/google/showcase/v1beta1/resumable_upload_service/rest.rb
index c7726e79c..a04e49147 100644
--- a/shared/output/gapic/templates/showcase/lib/google/showcase/v1beta1/resumable_upload_service/rest.rb
+++ b/shared/output/gapic/templates/showcase/lib/google/showcase/v1beta1/resumable_upload_service/rest.rb
@@ -32,6 +32,7 @@
 require "google/showcase/v1beta1/bindings_override"
 
 require "google/showcase/v1beta1/resumable_upload_service/credentials"
+require "google/showcase/v1beta1/resumable_upload_service/resumable_upload_stub"
 require "google/showcase/v1beta1/resumable_upload_service/rest/client"
 
 module Google
diff --git a/shared/output/gapic/templates/showcase/lib/google/showcase/v1beta1/resumable_upload_service/rest/client.rb b/shared/output/gapic/templates/showcase/lib/google/showcase/v1beta1/resumable_upload_service/rest/client.rb
index 3f9605149..7d9e2be58 100644
--- a/shared/output/gapic/templates/showcase/lib/google/showcase/v1beta1/resumable_upload_service/rest/client.rb
+++ b/shared/output/gapic/templates/showcase/lib/google/showcase/v1beta1/resumable_upload_service/rest/client.rb
@@ -39,6 +39,10 @@ module Rest
           #
           # A service showcasing universal resumable upload protocol support.
           #
+          # `upload_media` performs a resumable upload: it returns a {::Gapic::ResumableUpload} handle
+          # instead of a response, and the per-call `timeout` and retry policy cover only the request that
+          # creates the upload session, not the upload itself.
+          #
           class Client
             # @private
             API_VERSION = ""
@@ -145,6 +149,14 @@ def initialize
               @quota_project_id = @config.quota_project
               @quota_project_id ||= credentials.quota_project_id if credentials.respond_to? :quota_project_id
 
+              @resumable_upload_stub = ::Google::Showcase::V1beta1::ResumableUploadService::ResumableUploadStub.new(
+                endpoint: @config.endpoint,
+                endpoint_template: DEFAULT_ENDPOINT_TEMPLATE,
+                universe_domain: @config.universe_domain,
+                credentials: credentials,
+                logger: @config.logger
+              )
+
               @resumable_upload_service_stub = ::Google::Showcase::V1beta1::ResumableUploadService::Rest::ServiceStub.new(
                 endpoint: @config.endpoint,
                 endpoint_template: DEFAULT_ENDPOINT_TEMPLATE,
@@ -210,15 +222,22 @@ def logger
             ##
             # A method with media_upload annotation enabled.
             #
-            # @overload upload_media(request, options = nil)
+            # @overload upload_media(request = {}, options = nil)
             #   Pass arguments to `upload_media` via a request object, either of type
             #   {::Google::Showcase::V1beta1::UploadMediaRequest} or an equivalent Hash.
             #
             #   @param request [::Google::Showcase::V1beta1::UploadMediaRequest, ::Hash]
-            #     A request object representing the call parameters. Required. To specify no
-            #     parameters, or to keep all the default parameter values, pass an empty Hash.
+            #     A request object representing the call parameters. Optional: it is ignored when the returned
+            #     handle is resumed rather than started, because a resumed upload targets a session the server
+            #     has already created.
             #   @param options [::Gapic::CallOptions, ::Hash]
-            #     Overrides the default settings for this call, e.g, timeout, retries etc. Optional.
+            #     Overrides for the initiation request only. The `timeout`, `retry_policy` and `metadata` set
+            #     here apply to the single request that creates the upload session, not to the upload as a
+            #     whole: an upload still transferring bytes an hour later has long outlived this `timeout`.
+            #     To bound the whole upload, pass `upload_timeout:` to {::Gapic::ResumableUpload#start} or
+            #     {::Gapic::ResumableUpload#resume}. Chunk transfers are retried by the upload protocol itself,
+            #     with the defaults documented on {::Gapic::ResumableUpload}; no call option reaches them.
+            #     Optional.
             #
             # @overload upload_media(name: nil)
             #   Pass arguments to `upload_media` via keyword arguments. Note that at
@@ -226,11 +245,9 @@ def logger
             #   the default parameter values, pass an empty Hash as a request object (see above).
             #
             #   @param name [::String]
-            # @yield [result, operation] Access the result along with the TransportOperation object
-            # @yieldparam result [::Google::Showcase::V1beta1::UploadMediaResponse]
-            # @yieldparam operation [::Gapic::Rest::TransportOperation]
-            #
-            # @return [::Google::Showcase::V1beta1::UploadMediaResponse]
+            # @return [::Gapic::ResumableUpload]
+            #   A reusable upload handle. No request is sent and no byte is read from a stream until
+            #   {::Gapic::ResumableUpload#start} or {::Gapic::ResumableUpload#resume} is called on it.
             #
             # @raise [::Gapic::Rest::Error] if the REST call is aborted.
             #
@@ -244,12 +261,18 @@ def logger
             #   request = Google::Showcase::V1beta1::UploadMediaRequest.new
             #
             #   # Call the upload_media method.
-            #   result = client.upload_media request
+            #   upload = client.upload_media request
+            #
+            #   # The returned object is a handle for a resumable upload. Nothing has been
+            #   # uploaded yet, and the timeout and retry policy of the call above cover only
+            #   # the request that creates the upload session, not the upload as a whole.
+            #   stream = File.open "input.bin", "rb"
+            #   result = upload.start stream: stream, content_type: "application/octet-stream"
             #
             #   # The returned object is of type Google::Showcase::V1beta1::UploadMediaResponse.
             #   p result
             #
-            def upload_media request, options = nil
+            def upload_media request = {}, options = nil
               raise ::ArgumentError, "request must be provided" if request.nil?
 
               request = ::Gapic::Protobuf.coerce request, to: ::Google::Showcase::V1beta1::UploadMediaRequest
@@ -277,11 +300,17 @@ def upload_media request, options = nil
                                      metadata:     @config.metadata,
                                      retry_policy: @config.retry_policy
 
-              @resumable_upload_service_stub.upload_media request, options do |result, operation|
-                yield result, operation if block_given?
-              end
-            rescue ::Faraday::Error => e
-              raise ::Gapic::Rest::Error.wrap_faraday_error e
+              ::Gapic::ResumableUpload.new(
+                client_stub_proc:     -> { @resumable_upload_stub.client_stub },
+                initial_request_proc: lambda {
+                  ::Google::Showcase::V1beta1::ResumableUploadService::ResumableUploadStub.transcode_upload_media_request request
+                },
+                initial_headers:      options.metadata,
+                start_retry_policy:   ::Gapic::Rest::ResumableUpload.start_retry_policy_for(options),
+                response_type:        ::Google::Showcase::V1beta1::UploadMediaResponse,
+                method_name:          "upload_media",
+                error_handler:        nil
+              )
             end
 
             ##
@@ -442,6 +471,12 @@ def rpcs
               class Rpcs
                 ##
                 # RPC-specific configuration for `upload_media`
+                #
+                # `upload_media` performs a resumable upload, so the `timeout` and `retry_policy`
+                # configured here cover the request that creates the upload session only, not the upload
+                # itself. The whole-upload budget is the `upload_timeout:` argument of
+                # {::Gapic::ResumableUpload#start} and {::Gapic::ResumableUpload#resume}.
+                #
                 # @return [::Gapic::Config::Method]
                 #
                 attr_reader :upload_media
diff --git a/shared/output/gapic/templates/showcase/lib/google/showcase/v1beta1/resumable_upload_service/rest/service_stub.rb b/shared/output/gapic/templates/showcase/lib/google/showcase/v1beta1/resumable_upload_service/rest/service_stub.rb
index b2b277453..48fc8b087 100644
--- a/shared/output/gapic/templates/showcase/lib/google/showcase/v1beta1/resumable_upload_service/rest/service_stub.rb
+++ b/shared/output/gapic/templates/showcase/lib/google/showcase/v1beta1/resumable_upload_service/rest/service_stub.rb
@@ -79,67 +79,6 @@ def endpoint
             def logger stub: false
               stub ? @client_stub.stub_logger : @client_stub.logger
             end
-
-            ##
-            # Baseline implementation for the upload_media REST call
-            #
-            # @param request_pb [::Google::Showcase::V1beta1::UploadMediaRequest]
-            #   A request object representing the call parameters. Required.
-            # @param options [::Gapic::CallOptions]
-            #   Overrides the default settings for this call, e.g, timeout, retries etc. Optional.
-            #
-            # @yield [result, operation] Access the result along with the TransportOperation object
-            # @yieldparam result [::Google::Showcase::V1beta1::UploadMediaResponse]
-            # @yieldparam operation [::Gapic::Rest::TransportOperation]
-            #
-            # @return [::Google::Showcase::V1beta1::UploadMediaResponse]
-            #   A result object deserialized from the server's reply
-            def upload_media request_pb, options = nil
-              raise ::ArgumentError, "request must be provided" if request_pb.nil?
-
-              verb, uri, query_string_params, body = ServiceStub.transcode_upload_media_request request_pb
-              query_string_params = if query_string_params.any?
-                                      query_string_params.to_h { |p| p.split "=", 2 }
-                                    else
-                                      {}
-                                    end
-
-              response = @client_stub.make_http_request(
-                verb,
-                uri: uri,
-                body: body || "",
-                params: query_string_params,
-                method_name: "upload_media",
-                options: options
-              )
-              operation = ::Gapic::Rest::TransportOperation.new response
-              result = ::Google::Showcase::V1beta1::UploadMediaResponse.decode_json response.body,
-                                                                                    ignore_unknown_fields: true
-              catch :response do
-                yield result, operation if block_given?
-                result
-              end
-            end
-
-            ##
-            # @private
-            #
-            # GRPC transcoding helper method for the upload_media REST call
-            #
-            # @param request_pb [::Google::Showcase::V1beta1::UploadMediaRequest]
-            #   A request object representing the call parameters. Required.
-            # @return [Array(String, [String, nil], Hash{String => String})]
-            #   Uri, Body, Query string parameters
-            def self.transcode_upload_media_request request_pb
-              transcoder = Gapic::Rest::GrpcTranscoder.new
-                                                      .with_bindings(
-                                                        uri_method: :post,
-                                                        uri_template: "/v1beta1/files:upload",
-                                                        body: "*",
-                                                        matches: []
-                                                      )
-              transcoder.transcode request_pb
-            end
           end
         end
       end
diff --git a/shared/output/gapic/templates/showcase/lib/google/showcase/v1beta1/resumable_upload_service/resumable_upload_stub.rb b/shared/output/gapic/templates/showcase/lib/google/showcase/v1beta1/resumable_upload_service/resumable_upload_stub.rb
new file mode 100644
index 000000000..d993845e3
--- /dev/null
+++ b/shared/output/gapic/templates/showcase/lib/google/showcase/v1beta1/resumable_upload_service/resumable_upload_stub.rb
@@ -0,0 +1,119 @@
+# frozen_string_literal: true
+
+# The MIT License (MIT)
+#
+# Copyright <YEAR> <COPYRIGHT HOLDER>
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in
+# all copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+# THE SOFTWARE.
+
+# Auto-generated by gapic-generator-ruby. DO NOT EDIT!
+
+
+module Google
+  module Showcase
+    module V1beta1
+      module ResumableUploadService
+        ##
+        # Resumable upload stub for the ResumableUploadService service.
+        #
+        # Owns the REST client stub the uploads are performed through, the transcoders that build their
+        # initiation requests, and the credentials error that a client unable to perform REST calls has to
+        # raise when an upload is attempted.
+        #
+        # @private
+        #
+        class ResumableUploadStub
+          # @private
+          UPLOAD_MEDIA_URL_PREFIX = "resumable/upload"
+
+          # @private
+          def initialize endpoint:, endpoint_template:, universe_domain:, credentials:, logger:
+            # These require statements are intentionally placed here to initialize
+            # the REST modules only when it's required.
+            require "gapic/rest"
+
+            # Uploads travel over REST, and credentials cannot be recovered from a gRPC channel. Rather than
+            # raise while a client is being constructed, which would break every other method on it, the
+            # error is stored and raised when an upload is actually attempted.
+            if defined?(::GRPC::Core::Channel) &&
+               (credentials.is_a?(::GRPC::Core::Channel) || credentials.is_a?(::GRPC::Core::ChannelCredentials))
+              @credentials_error = ::ArgumentError.new(
+                "Resumable uploads are performed over REST and cannot use a gRPC channel as credentials, " \
+                "because credentials cannot be recovered from a channel. Construct the client with Google " \
+                "OAuth credentials (a keyfile path, a Hash, or a Google::Auth::Credentials instance) to use " \
+                "resumable upload methods."
+              )
+              return
+            end
+
+            @client_stub = ::Gapic::Rest::ClientStub.new endpoint: endpoint,
+                                                         endpoint_template: endpoint_template,
+                                                         universe_domain: universe_domain,
+                                                         credentials: credentials,
+                                                         numeric_enums: false,
+                                                         service_name: self.class,
+                                                         raise_faraday_errors: false,
+                                                         logger: logger
+          end
+
+          ##
+          # @private
+          #
+          # The REST client stub the uploads are performed through.
+          #
+          # @return [::Gapic::Rest::ClientStub]
+          #
+          # @raise [::ArgumentError] if this client cannot perform REST calls.
+          #
+          def client_stub
+            raise @credentials_error if @credentials_error
+            @client_stub
+          end
+
+          ##
+          # @private
+          #
+          # GRPC transcoding helper method for the upload_media resumable upload initiation request.
+          #
+          # The upload itself is not transcoded: only the request that creates the upload session is, and the
+          # verb is discarded because initiation is always a POST. Query string parameters are folded into the
+          # URL, since the upload sends the initiation request with no separate parameters.
+          #
+          # @param request_pb [::Google::Showcase::V1beta1::UploadMediaRequest]
+          #   A request object representing the call parameters. Required.
+          # @return [Array(String, [String, nil])]
+          #   Uri, Body
+          def self.transcode_upload_media_request request_pb
+            transcoder = ::Gapic::Rest::GrpcTranscoder.new
+                                                      .with_bindings(
+                                                        uri_method: :post,
+                                                        uri_template: "/v1beta1/files:upload",
+                                                        body: "*",
+                                                        matches: []
+                                                      )
+            _verb, uri, query_string_params, body = transcoder.transcode request_pb
+            uri = "/#{UPLOAD_MEDIA_URL_PREFIX}#{uri}"
+            uri = "#{uri}?#{query_string_params.join '&'}" if query_string_params.any?
+            [uri, body]
+          end
+        end
+      end
+    end
+  end
+end
diff --git a/shared/output/gapic/templates/showcase/snippets/resumable_upload_service/upload_media.rb b/shared/output/gapic/templates/showcase/snippets/resumable_upload_service/upload_media.rb
index c16a80be8..099f8d7ba 100644
--- a/shared/output/gapic/templates/showcase/snippets/resumable_upload_service/upload_media.rb
+++ b/shared/output/gapic/templates/showcase/snippets/resumable_upload_service/upload_media.rb
@@ -47,7 +47,13 @@ def upload_media
   request = Google::Showcase::V1beta1::UploadMediaRequest.new
 
   # Call the upload_media method.
-  result = client.upload_media request
+  upload = client.upload_media request
+
+  # The returned object is a handle for a resumable upload. Nothing has been
+  # uploaded yet, and the timeout and retry policy of the call above cover only
+  # the request that creates the upload session, not the upload as a whole.
+  stream = File.open "input.bin", "rb"
+  result = upload.start stream: stream, content_type: "application/octet-stream"
 
   # The returned object is of type Google::Showcase::V1beta1::UploadMediaResponse.
   p result
diff --git a/shared/output/gapic/templates/showcase/snippets/snippet_metadata_google.showcase.v1beta1.json b/shared/output/gapic/templates/showcase/snippets/snippet_metadata_google.showcase.v1beta1.json
index 8679f7e93..d73f66677 100644
--- a/shared/output/gapic/templates/showcase/snippets/snippet_metadata_google.showcase.v1beta1.json
+++ b/shared/output/gapic/templates/showcase/snippets/snippet_metadata_google.showcase.v1beta1.json
@@ -1646,7 +1646,7 @@
       "segments": [
         {
           "start": 28,
-          "end": 54,
+          "end": 60,
           "type": "FULL"
         }
       ]
diff --git a/shared/output/gapic/templates/showcase/test/google/showcase/v1beta1/resumable_upload_service_rest_test.rb b/shared/output/gapic/templates/showcase/test/google/showcase/v1beta1/resumable_upload_service_rest_test.rb
index ec3598336..6d9f676b4 100644
--- a/shared/output/gapic/templates/showcase/test/google/showcase/v1beta1/resumable_upload_service_rest_test.rb
+++ b/shared/output/gapic/templates/showcase/test/google/showcase/v1beta1/resumable_upload_service_rest_test.rb
@@ -86,60 +86,6 @@ def logger
     end
   end
 
-  def test_upload_media
-    # Create test objects.
-    client_result = ::Google::Showcase::V1beta1::UploadMediaResponse.new
-    http_response = OpenStruct.new body: client_result.to_json
-
-    call_options = {}
-
-    # Create request parameters for a unary method.
-    name = "hello world"
-
-    upload_media_client_stub = ClientStub.new http_response do |_verb, uri:, body:, params:, options:, method_name:|
-      assert options.metadata.key? :"x-goog-api-client"
-      assert options.metadata[:"x-goog-api-client"].include? "rest"
-      refute options.metadata[:"x-goog-api-client"].include? "grpc"
-    end
-
-    ::Google::Showcase::V1beta1::ResumableUploadService::Rest::ServiceStub.stub :transcode_upload_media_request, ["", "", {}] do
-      Gapic::Rest::ClientStub.stub :new, upload_media_client_stub do
-        # Create client
-        c = ::Google::Showcase::V1beta1::ResumableUploadService::Rest::Client.new do |config|
-          config.credentials = :dummy_value
-        end
-
-        # Use hash object
-        c.upload_media({ name: name }) do |_result, response|
-          assert_equal http_response, response.underlying_op
-        end
-
-        # Use named arguments
-        c.upload_media name: name do |_result, response|
-          assert_equal http_response, response.underlying_op
-        end
-
-        # Use protobuf object
-        c.upload_media ::Google::Showcase::V1beta1::UploadMediaRequest.new(name: name) do |_result, response|
-          assert_equal http_response, response.underlying_op
-        end
-
-        # Use hash object with options
-        c.upload_media({ name: name }, call_options) do |_result, response|
-          assert_equal http_response, response.underlying_op
-        end
-
-        # Use protobuf object with options
-        c.upload_media(::Google::Showcase::V1beta1::UploadMediaRequest.new(name: name), call_options) do |_result, response|
-          assert_equal http_response, response.underlying_op
-        end
-
-        # Verify method calls
-        assert_equal 5, upload_media_client_stub.call_count
-      end
-    end
-  end
-
   def test_configure
     credentials_token = :dummy_value
 
diff --git a/shared/output/gapic/templates/showcase/test/google/showcase/v1beta1/resumable_upload_service_resumable_upload_test.rb b/shared/output/gapic/templates/showcase/test/google/showcase/v1beta1/resumable_upload_service_resumable_upload_test.rb
new file mode 100644
index 000000000..9ca701619
--- /dev/null
+++ b/shared/output/gapic/templates/showcase/test/google/showcase/v1beta1/resumable_upload_service_resumable_upload_test.rb
@@ -0,0 +1,136 @@
+# frozen_string_literal: true
+
+# The MIT License (MIT)
+#
+# Copyright <YEAR> <COPYRIGHT HOLDER>
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in
+# all copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+# THE SOFTWARE.
+
+# Auto-generated by gapic-generator-ruby. DO NOT EDIT!
+
+require "helper"
+
+require "gapic/rest"
+require "gapic/grpc/service_stub"
+
+require "google/showcase/v1beta1/resumable_upload_pb"
+require "google/showcase/v1beta1/resumable_upload_service"
+require "google/showcase/v1beta1/resumable_upload_service/rest"
+
+class ::Google::Showcase::V1beta1::ResumableUploadService::ResumableUploadTest < Minitest::Test
+  ##
+  # Stands in for the transport stub a client builds in its constructor. An upload method sends
+  # nothing, so this only has to satisfy the constructor.
+  #
+  class ClientStub
+    def endpoint
+      "endpoint.example.com"
+    end
+
+    def universe_domain
+      "example.com"
+    end
+
+    def stub_logger
+      nil
+    end
+
+    def logger
+      nil
+    end
+  end
+
+  def test_transcode_upload_media_request
+    request = ::Google::Showcase::V1beta1::UploadMediaRequest.new
+
+    # The transcoder is stubbed so the test pins the upload-specific post-processing (prefix,
+    # query-string folding, [uri, body] shape) independently of the method's path template.
+    transcoder = Gapic::Rest::GrpcTranscoder.new
+    transcoder.stub :transcode, [:post, "/transcoded/path", ["foo=bar", "baz=qux"], "{\"body\":true}"] do
+      Gapic::Rest::GrpcTranscoder.stub :new, transcoder do
+        uri, body = ::Google::Showcase::V1beta1::ResumableUploadService::ResumableUploadStub.transcode_upload_media_request request
+
+        assert_equal "/#{::Google::Showcase::V1beta1::ResumableUploadService::ResumableUploadStub::UPLOAD_MEDIA_URL_PREFIX}/transcoded/path?foo=bar&baz=qux", uri
+        assert_equal "{\"body\":true}", body
+      end
+    end
+  end
+
+  def test_upload_media
+    name = "hello world"
+
+    client = nil
+    Gapic::ServiceStub.stub :new, ClientStub.new do
+      client = ::Google::Showcase::V1beta1::ResumableUploadService::Client.new do |config|
+        config.credentials = :this_channel_is_insecure
+      end
+    end
+
+    upload = client.upload_media ::Google::Showcase::V1beta1::UploadMediaRequest.new(name: name)
+
+    # The call builds a handle and sends nothing. The upload only begins on #start or #resume.
+    assert_kind_of ::Gapic::ResumableUpload, upload
+    assert_nil upload.resume_handle
+    refute upload.resumable?
+    refute upload.running?
+  end
+
+  def test_upload_media_channel_credentials
+    name = "hello world"
+    grpc_channel = GRPC::Core::Channel.new "localhost:8888", nil, :this_channel_is_insecure
+
+    client = nil
+    Gapic::ServiceStub.stub :new, ClientStub.new do
+      client = ::Google::Showcase::V1beta1::ResumableUploadService::Client.new do |config|
+        config.credentials = grpc_channel
+      end
+    end
+
+    upload = client.upload_media ::Google::Showcase::V1beta1::UploadMediaRequest.new(name: name)
+
+    assert_kind_of ::Gapic::ResumableUpload, upload
+
+    # Uploads travel over REST and cannot use a channel as credentials. The error surfaces here,
+    # rather than at construction, and before the stream is touched.
+    stream = StringIO.new "resumable upload test payload"
+    assert_raises ::ArgumentError do
+      upload.start stream: stream, content_type: "application/octet-stream"
+    end
+    assert_equal 0, stream.pos
+  end
+
+  def test_upload_media_rest
+    name = "hello world"
+
+    client = nil
+    Gapic::Rest::ClientStub.stub :new, ClientStub.new do
+      client = ::Google::Showcase::V1beta1::ResumableUploadService::Rest::Client.new do |config|
+        config.credentials = :dummy_value
+      end
+    end
+
+    upload = client.upload_media ::Google::Showcase::V1beta1::UploadMediaRequest.new(name: name)
+
+    # The call builds a handle and sends nothing. The upload only begins on #start or #resume.
+    assert_kind_of ::Gapic::ResumableUpload, upload
+    assert_nil upload.resume_handle
+    refute upload.resumable?
+    refute upload.running?
+  end
+end
diff --git a/shared/output/gapic/templates/showcase/test/google/showcase/v1beta1/resumable_upload_service_test.rb b/shared/output/gapic/templates/showcase/test/google/showcase/v1beta1/resumable_upload_service_test.rb
index 9538f0539..f9fbf3269 100644
--- a/shared/output/gapic/templates/showcase/test/google/showcase/v1beta1/resumable_upload_service_test.rb
+++ b/shared/output/gapic/templates/showcase/test/google/showcase/v1beta1/resumable_upload_service_test.rb
@@ -71,64 +71,6 @@ def logger
     end
   end
 
-  def test_upload_media
-    # Create GRPC objects.
-    grpc_response = ::Google::Showcase::V1beta1::UploadMediaResponse.new
-    grpc_operation = GRPC::ActiveCall::Operation.new nil
-    grpc_channel = GRPC::Core::Channel.new "localhost:8888", nil, :this_channel_is_insecure
-    grpc_options = {}
-
-    # Create request parameters for a unary method.
-    name = "hello world"
-
-    upload_media_client_stub = ClientStub.new grpc_response, grpc_operation do |name, request, options:|
-      assert_equal :upload_media, name
-      assert_kind_of ::Google::Showcase::V1beta1::UploadMediaRequest, request
-      assert_equal "hello world", request["name"]
-      refute_nil options
-    end
-
-    Gapic::ServiceStub.stub :new, upload_media_client_stub do
-      # Create client
-      c = ::Google::Showcase::V1beta1::ResumableUploadService::Client.new do |config|
-        config.credentials = grpc_channel
-      end
-
-      # Use hash object
-      c.upload_media({ name: name }) do |response, operation|
-        assert_equal grpc_response, response
-        assert_equal grpc_operation, operation
-      end
-
-      # Use named arguments
-      c.upload_media name: name do |response, operation|
-        assert_equal grpc_response, response
-        assert_equal grpc_operation, operation
-      end
-
-      # Use protobuf object
-      c.upload_media ::Google::Showcase::V1beta1::UploadMediaRequest.new(name: name) do |response, operation|
-        assert_equal grpc_response, response
-        assert_equal grpc_operation, operation
-      end
-
-      # Use hash object with options
-      c.upload_media({ name: name }, grpc_options) do |response, operation|
-        assert_equal grpc_response, response
-        assert_equal grpc_operation, operation
-      end
-
-      # Use protobuf object with options
-      c.upload_media(::Google::Showcase::V1beta1::UploadMediaRequest.new(name: name), grpc_options) do |response, operation|
-        assert_equal grpc_response, response
-        assert_equal grpc_operation, operation
-      end
-
-      # Verify method calls
-      assert_equal 5, upload_media_client_stub.call_rpc_count
-    end
-  end
-
   def test_configure
     grpc_channel = GRPC::Core::Channel.new "localhost:8888", nil, :this_channel_is_insecure