diff --git a/google/cloud/storage/internal/async/multi_stream_manager.h b/google/cloud/storage/internal/async/multi_stream_manager.h
index bf26914127b88..876f228d5ada3 100644
--- a/google/cloud/storage/internal/async/multi_stream_manager.h
+++ b/google/cloud/storage/internal/async/multi_stream_manager.h
@@ -19,6 +19,7 @@
 #include "google/cloud/version.h"
 #include <cstdint>
 #include <functional>
+#include <limits>
 #include <list>
 #include <memory>
 #include <unordered_map>
@@ -89,18 +90,21 @@ class MultiStreamManager {
     return streams_.begin();
   }
 
-  StreamIterator GetLeastBusyStream() {
+  // Returns an iterator to the stream with the fewest active ranges matching
+  // the given predicate. Returns End() if no stream satisfies the predicate.
+  // Strict less-than ensures stability by preferring earlier (older) streams if
+  // tied.
+  template <typename Pred>
+  StreamIterator GetLeastBusyStream(Pred pred) {
     if (streams_.empty()) return streams_.end();
-    auto least_busy_stream_it = streams_.begin();
-    // Track min_ranges to avoid calling .size() repeatedly if possible,
-    // though for std::unordered_map .size() is O(1).
-    std::size_t min_ranges = least_busy_stream_it->active_ranges.size();
-    if (min_ranges == 0) return least_busy_stream_it;
-
-    // Start checking from the second element
-    for (auto it = std::next(streams_.begin()); it != streams_.end(); ++it) {
-      // Strict less-than ensures stability (preferring older streams if tied)
-      auto size = it->active_ranges.size();
+    auto least_busy_stream_it = streams_.end();
+    // Track min_ranges to avoid calling .size() repeatedly.
+    std::size_t min_ranges = (std::numeric_limits<std::size_t>::max)();
+
+    for (auto it = streams_.begin(); it != streams_.end(); ++it) {
+      if (!pred(*it)) continue;
+      std::size_t const size = it->active_ranges.size();
+      // Strict less-than ensures stability (preferring older streams if tied).
       if (size < min_ranges) {
         least_busy_stream_it = it;
         min_ranges = size;
@@ -110,6 +114,12 @@ class MultiStreamManager {
     return least_busy_stream_it;
   }
 
+  // Overload of `GetLeastBusyStream` without predicate that selects the stream
+  // with the fewest active ranges across all managed streams.
+  StreamIterator GetLeastBusyStream() {
+    return GetLeastBusyStream([](Stream const&) { return true; });
+  }
+
   StreamIterator AddStream(std::shared_ptr<StreamT> stream) {
     streams_.emplace_front(Stream{std::move(stream), {}});
     return streams_.begin();
@@ -159,6 +169,13 @@ class MultiStreamManager {
     return false;
   }
 
+  StreamIterator Find(std::shared_ptr<StreamT> const& target) {
+    for (auto it = streams_.begin(); it != streams_.end(); ++it) {
+      if (it->stream == target) return it;
+    }
+    return streams_.end();
+  }
+
   bool Empty() const { return streams_.empty(); }
   ConstStreamIterator End() const { return streams_.end(); }
   std::size_t Size() const { return streams_.size(); }
diff --git a/google/cloud/storage/internal/async/multi_stream_manager_test.cc b/google/cloud/storage/internal/async/multi_stream_manager_test.cc
index b17fa4d32173a..d3dd69c9ef3da 100644
--- a/google/cloud/storage/internal/async/multi_stream_manager_test.cc
+++ b/google/cloud/storage/internal/async/multi_stream_manager_test.cc
@@ -101,6 +101,37 @@ TEST(MultiStreamManagerTest, GetLeastBusyPrefersFewestActiveRanges) {
   EXPECT_EQ(it_least->active_ranges.size(), 1U);
 }
 
+/// @test Verifies that GetLeastBusyStream with a predicate filters candidates
+/// based on the provided predicate, correctly returning End() if no stream
+/// matches or the least busy stream among those that satisfy the predicate.
+TEST(MultiStreamManagerTest, GetLeastBusyStreamWithPredicate) {
+  auto mgr = MultiStreamManagerTest::MakeManager();
+  mgr.GetFirstStream()->stream->write_pending = true;
+
+  auto s1 = std::make_shared<FakeStream>();
+  auto s2 = std::make_shared<FakeStream>();
+  mgr.AddStream(s1);
+  auto it2 = mgr.AddStream(s2);
+
+  // s1 has 0 ranges, but write_pending = true.
+  s1->write_pending = true;
+  // s2 has 1 range, write_pending = false.
+  s2->write_pending = false;
+  it2->active_ranges.emplace(1, std::make_shared<FakeRange>());
+
+  // Predicate filtering out write_pending streams selects s2 even though it has
+  // more ranges than s1.
+  auto it_pred = mgr.GetLeastBusyStream([](Manager::Stream const& s) {
+    return s.stream != nullptr && !s.stream->write_pending;
+  });
+  EXPECT_THAT(it_pred, ::testing::Eq(it2));
+
+  // If predicate matches no stream, returns End().
+  auto it_none =
+      mgr.GetLeastBusyStream([](Manager::Stream const&) { return false; });
+  EXPECT_THAT(it_none, ::testing::Eq(mgr.End()));
+}
+
 TEST(MultiStreamManagerTest, CleanupDoneRangesRemovesFinished) {
   auto mgr = MultiStreamManagerTest::MakeManager();
   auto it = mgr.GetFirstStream();
@@ -233,6 +264,22 @@ TEST(MultiStreamManagerTest, EmptyAndSizeTransitions) {
   EXPECT_EQ(mgr.Size(), 1U);
 }
 
+TEST(MultiStreamManagerTest, FindTracksStreamMembership) {
+  auto mgr = MultiStreamManagerTest::MakeManager();
+  auto it1 = mgr.GetFirstStream();
+  auto s1 = it1->stream;
+  EXPECT_NE(mgr.Find(s1), mgr.End());
+
+  auto s2 = std::make_shared<FakeStream>();
+  mgr.AddStream(s2);
+  EXPECT_NE(mgr.Find(s1), mgr.End());
+  EXPECT_NE(mgr.Find(s2), mgr.End());
+
+  mgr.RemoveStreamAndNotifyRanges(it1, Status());
+  EXPECT_EQ(mgr.Find(s1), mgr.End());
+  EXPECT_NE(mgr.Find(s2), mgr.End());
+}
+
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
 }  // namespace storage_internal
 }  // namespace cloud
diff --git a/google/cloud/storage/internal/async/object_descriptor_impl.cc b/google/cloud/storage/internal/async/object_descriptor_impl.cc
index 194f288c66744..4e06e316e24e0 100644
--- a/google/cloud/storage/internal/async/object_descriptor_impl.cc
+++ b/google/cloud/storage/internal/async/object_descriptor_impl.cc
@@ -77,10 +77,14 @@ ObjectDescriptorImpl::ObjectDescriptorImpl(
       options_(std::move(options)),
       has_initial_read_ranges_(options_.has<ReadRangesOption>()),
       transport_ok_(std::move(transport_ok)) {
+  auto initial_read_stream = std::make_shared<ReadStream>(
+      std::move(stream), resume_policy_prototype_->clone());
+  // Notify the resume policy that the initial stream was established
+  // successfully.
+  initial_read_stream->resume_policy->OnStartSuccess();
   stream_manager_ = std::make_unique<StreamManager>(
       []() -> std::shared_ptr<ReadStream> { return nullptr; },  // NOLINT
-      std::make_shared<ReadStream>(std::move(stream),
-                                   resume_policy_prototype_->clone()));
+      std::move(initial_read_stream));
   // Initialize the pacing limit from options if configured.
   if (options_.has<PreWarmBufferLimitOption>()) {
     max_prewarmed_buffer_size_ = options_.get<PreWarmBufferLimitOption>();
@@ -122,8 +126,10 @@ void ObjectDescriptorImpl::Start(
   std::unique_lock<std::mutex> lk(mu_);
   auto it = stream_manager_->GetFirstStream();
   if (it == stream_manager_->End()) return;
+  std::shared_ptr<ReadStream> read_stream = it->stream;
+  std::shared_ptr<OpenStream> current_stream = read_stream->stream;
   lk.unlock();
-  OnRead(it, std::move(first_response));
+  OnRead(read_stream, current_stream, std::move(first_response));
   // Acquire lock and queue the background stream if multi-stream optimization
   // is enabled.
   if (options_.get<storage::EnableMultiStreamOptimizationOption>()) {
@@ -171,12 +177,14 @@ void ObjectDescriptorImpl::MakeSubsequentStream() {
   }
 
   std::unique_lock<std::mutex> lk(mu_);
-  // Reuse an idle stream if possible.
+  // Reuse an idle stream if possible. A stream undergoing reconnection
+  // (resuming == true) must not be treated as idle, as it cannot accept new
+  // read ranges immediately.
   if (stream_manager_->ReuseIdleStreamToFront(
           [](StreamManager::Stream const& s) {
             auto const* rs = s.stream.get();
             return rs != nullptr && s.active_ranges.empty() &&
-                   !rs->write_pending;
+                   !rs->write_pending && !rs->resuming;
           })) {
     return;
   }
@@ -201,17 +209,20 @@ void ObjectDescriptorImpl::MakeSubsequentStream() {
     std::unique_lock<std::mutex> lk(self->mu_);
     if (self->cancelled_) return;
 
-    auto read_stream =
+    std::shared_ptr<ReadStream> read_stream =
         std::make_shared<ReadStream>(std::move(stream_result->stream),
                                      self->resume_policy_prototype_->clone());
+    read_stream->resume_policy->OnStartSuccess();
 
-    auto new_it = self->stream_manager_->AddStream(std::move(read_stream));
+    self->stream_manager_->AddStream(read_stream);
 
     // Now that we consumed pending_stream_, queue the next one immediately.
     self->AssurePendingStreamQueued(lk);
 
+    std::shared_ptr<OpenStream> new_stream = read_stream->stream;
     lk.unlock();
-    self->OnRead(new_it, std::move(stream_result->first_response));
+    self->OnRead(read_stream, new_stream,
+                 std::move(stream_result->first_response));
   });
 }
 
@@ -307,14 +318,26 @@ std::unique_ptr<storage::AsyncReaderConnection> ObjectDescriptorImpl::Read(
                                              CacheStatusToString(cache_status));
   }
 
-  auto it = stream_manager_->GetLeastBusyStream();
-  auto const id = ++read_id_generator_;
+  // Prioritize selecting a healthy stream that is not undergoing reconnection.
+  // If all streams are currently reconnecting, fall back to the least busy
+  // resuming stream so that the range is queued in next_request and dispatched
+  // upon reconnection completion in OnResume().
+  auto it =
+      stream_manager_->GetLeastBusyStream([](StreamManager::Stream const& s) {
+        auto const* rs = s.stream.get();
+        return rs != nullptr && !rs->resuming;
+      });
+  if (it == stream_manager_->End()) {
+    it = stream_manager_->GetLeastBusyStream();
+  }
+  std::shared_ptr<ReadStream> read_stream = it->stream;
+  std::int64_t const id = ++read_id_generator_;
   it->active_ranges.emplace(id, range);
-  auto& read_range = *it->stream->next_request.add_read_ranges();
+  auto& read_range = *read_stream->next_request.add_read_ranges();
   read_range.set_read_id(id);
   read_range.set_read_offset(p.start);
   read_range.set_read_length(p.length);
-  Flush(std::move(lk), it);
+  Flush(std::move(lk), read_stream);
 
   if (!internal::TracingEnabled(options_)) {
     return std::unique_ptr<storage::AsyncReaderConnection>(
@@ -398,56 +421,80 @@ ObjectDescriptorImpl::CreateHashValidator(bool is_full_read) const {
   return hash_validator;
 }
 
-void ObjectDescriptorImpl::Flush(std::unique_lock<std::mutex> lk,
-                                 StreamIterator it) {
-  if (it->stream->write_pending ||
-      it->stream->next_request.read_ranges().empty()) {
+void ObjectDescriptorImpl::Flush(
+    std::unique_lock<std::mutex> lk,
+    std::shared_ptr<ReadStream> const& read_stream) {
+  if (!read_stream || read_stream->resuming || read_stream->write_pending ||
+      read_stream->next_request.read_ranges().empty()) {
     return;
   }
-  it->stream->write_pending = true;
+  read_stream->write_pending = true;
   google::storage::v2::BidiReadObjectRequest request;
-  request.Swap(&it->stream->next_request);
+  request.Swap(&read_stream->next_request);
 
   // Assign CurrentStream to a temporary variable to prevent
   // lifetime extension which can cause the lock to be held until the
   // end of the block.
-  auto current_stream = it->stream->stream;
+  std::shared_ptr<OpenStream> current_stream = read_stream->stream;
   lk.unlock();
   current_stream->Write(std::move(request))
-      .then([w = WeakFromThis(), it](auto f) {
-        if (auto self = w.lock()) self->OnWrite(it, f.get());
+      .then([w = WeakFromThis(), read_stream, current_stream](auto f) {
+        if (auto self = w.lock()) {
+          self->OnWrite(read_stream, current_stream, f.get());
+        }
       });
 }
 
-void ObjectDescriptorImpl::OnWrite(StreamIterator it, bool ok) {
+void ObjectDescriptorImpl::OnWrite(
+    std::shared_ptr<ReadStream> const& read_stream,
+    std::shared_ptr<OpenStream> const& stream, bool ok) {
   std::unique_lock<std::mutex> lk(mu_);
-  if (!ok) return DoFinish(std::move(lk), it);
+  // Discard callbacks from stale or removed streams (e.g. if the stream was
+  // replaced during reconnection or removed after an error).
+  auto it = stream_manager_->Find(read_stream);
+  if (it == stream_manager_->End() || !it->stream ||
+      it->stream->stream != stream) {
+    return;
+  }
+  if (!ok) return DoFinish(std::move(lk), read_stream, stream);
   it->stream->write_pending = false;
-  Flush(std::move(lk), it);
+  Flush(std::move(lk), read_stream);
 }
 
-void ObjectDescriptorImpl::DoRead(std::unique_lock<std::mutex> lk,
-                                  StreamIterator it) {
-  if (it->stream->read_pending) return;
-  it->stream->read_pending = true;
+void ObjectDescriptorImpl::DoRead(
+    std::unique_lock<std::mutex> lk,
+    std::shared_ptr<ReadStream> const& read_stream) {
+  if (!read_stream || read_stream->read_pending) return;
+  read_stream->read_pending = true;
 
   // Assign CurrentStream to a temporary variable to prevent
   // lifetime extension which can cause the lock to be held until the
   // end of the block.
-  auto current_stream = it->stream->stream;
+  std::shared_ptr<OpenStream> current_stream = read_stream->stream;
   lk.unlock();
-  current_stream->Read().then([w = WeakFromThis(), it](auto f) {
-    if (auto self = w.lock()) self->OnRead(it, f.get());
-  });
+  current_stream->Read().then(
+      [w = WeakFromThis(), read_stream, current_stream](auto f) {
+        if (auto self = w.lock()) {
+          self->OnRead(read_stream, current_stream, f.get());
+        }
+      });
 }
 
 void ObjectDescriptorImpl::OnRead(
-    StreamIterator it,
+    std::shared_ptr<ReadStream> const& read_stream,
+    std::shared_ptr<OpenStream> const& stream,
     std::optional<google::storage::v2::BidiReadObjectResponse> response) {
   std::unique_lock<std::mutex> lk(mu_);
+  // Discard callbacks from stale or removed streams (e.g. if the stream was
+  // replaced during reconnection or removed after an error).
+  auto it = stream_manager_->Find(read_stream);
+  if (it == stream_manager_->End() || !it->stream ||
+      it->stream->stream != stream) {
+    return;
+  }
   it->stream->read_pending = false;
 
-  if (!response) return DoFinish(std::move(lk), it);
+  if (!response) return DoFinish(std::move(lk), read_stream, stream);
   if (response->has_metadata()) {
     metadata_ = std::move(*response->mutable_metadata());
   }
@@ -465,52 +512,25 @@ void ObjectDescriptorImpl::OnRead(
   // Release the lock while notifying the ranges. The notifications may trigger
   // application code, and that code may callback on this class.
   lk.unlock();
-  auto apply_pacing_and_check_eviction = [this](std::int64_t id,
-                                                std::size_t chunk_size,
-                                                StreamIterator it) {
-    auto unclaimed_it = unclaimed_ranges_.find(id);
-    if (unclaimed_it == unclaimed_ranges_.end()) return false;
-
-    if (total_prewarmed_bytes_buffered_ + chunk_size >
-        max_prewarmed_buffer_size_) {
-      // Evict the range if it exceeds the pacing limit.
-      total_prewarmed_bytes_buffered_ -= unclaimed_it->second.bytes_buffered;
-      // Cap tombstone set size to prevent unbounded memory growth in long-lived
-      // descriptors where pre-warmed ranges are evicted but never requested.
-      if (evicted_ranges_.size() < 1000) {
-        evicted_ranges_.insert(unclaimed_it->second.cache_it->first);
-      }
-      prewarmed_ranges_.erase(unclaimed_it->second.cache_it);
-      unclaimed_ranges_.erase(unclaimed_it);
-
-      // Erasing from active_ranges ensures we ignore any subsequent GCS chunks
-      // for this range.
-      it->active_ranges.erase(id);
-      return true;
-    }
-
-    // Track buffered data size for pacing.
-    unclaimed_it->second.bytes_buffered += chunk_size;
-    total_prewarmed_bytes_buffered_ += chunk_size;
-    return false;
-  };
 
   for (auto& range_data : *response->mutable_object_data_ranges()) {
-    auto id = range_data.read_range().read_id();
+    std::int64_t id = range_data.read_range().read_id();
     auto const l = copy.find(id);
     if (l == copy.end()) continue;
 
     auto range = l->second;
-    auto chunk_size = range_data.checksummed_data().content().size();
+    std::size_t chunk_size = range_data.checksummed_data().content().size();
 
     bool evict = false;
     lk.lock();
     // Verify the range is still active under the lock. Because `OnRead`
     // processes chunks in batches, an earlier chunk in the same batch could
     // breach the pacing limit and evict a subsequent chunk's range.
-    bool active = it->active_ranges.count(id) != 0;
+    auto it_curr = stream_manager_->Find(read_stream);
+    bool active = (it_curr != stream_manager_->End()) &&
+                  (it_curr->active_ranges.count(id) != 0);
     if (active) {
-      evict = apply_pacing_and_check_eviction(id, chunk_size, it);
+      evict = ApplyPacingAndCheckEviction(id, chunk_size, it_curr);
     }
     lk.unlock();
     if (active) {
@@ -527,12 +547,52 @@ void ObjectDescriptorImpl::OnRead(
     }
   }
   lk.lock();
-  stream_manager_->CleanupDoneRanges(it);
-  DoRead(std::move(lk), it);
+  auto it_final = stream_manager_->Find(read_stream);
+  if (it_final == stream_manager_->End() || !it_final->stream) return;
+  stream_manager_->CleanupDoneRanges(it_final);
+  DoRead(std::move(lk), read_stream);
+}
+
+bool ObjectDescriptorImpl::ApplyPacingAndCheckEviction(std::int64_t id,
+                                                       std::size_t chunk_size,
+                                                       StreamIterator it) {
+  auto unclaimed_it = unclaimed_ranges_.find(id);
+  if (unclaimed_it == unclaimed_ranges_.end()) return false;
+
+  if (total_prewarmed_bytes_buffered_ + chunk_size >
+      max_prewarmed_buffer_size_) {
+    // Evict the range if it exceeds the pacing limit.
+    total_prewarmed_bytes_buffered_ -= unclaimed_it->second.bytes_buffered;
+    // Cap tombstone set size to prevent unbounded memory growth in long-lived
+    // descriptors where pre-warmed ranges are evicted but never requested.
+    if (evicted_ranges_.size() < 1000) {
+      evicted_ranges_.insert(unclaimed_it->second.cache_it->first);
+    }
+    prewarmed_ranges_.erase(unclaimed_it->second.cache_it);
+    unclaimed_ranges_.erase(unclaimed_it);
+
+    // Erasing from active_ranges ensures we ignore any subsequent GCS chunks
+    // for this range.
+    it->active_ranges.erase(id);
+    return true;
+  }
+
+  // Track buffered data size for pacing.
+  unclaimed_it->second.bytes_buffered += chunk_size;
+  total_prewarmed_bytes_buffered_ += chunk_size;
+  return false;
 }
 
-void ObjectDescriptorImpl::DoFinish(std::unique_lock<std::mutex> lk,
-                                    StreamIterator it) {
+void ObjectDescriptorImpl::DoFinish(
+    std::unique_lock<std::mutex> lk,
+    std::shared_ptr<ReadStream> const& read_stream,
+    std::shared_ptr<OpenStream> const& stream) {
+  // Discard finish requests if the stream was already replaced or removed.
+  auto it = stream_manager_->Find(read_stream);
+  if (it == stream_manager_->End() || !it->stream ||
+      it->stream->stream != stream) {
+    return;
+  }
   it->stream->read_pending = false;
   // Assign CurrentStream to a temporary variable to prevent
   // lifetime extension which can cause the lock to be held until the
@@ -541,24 +601,58 @@ void ObjectDescriptorImpl::DoFinish(std::unique_lock<std::mutex> lk,
   lk.unlock();
   auto pending = current_stream->Finish();
   if (!pending.valid()) return;
-  pending.then([w = WeakFromThis(), it](auto f) {
-    if (auto self = w.lock()) self->OnFinish(it, f.get());
+  pending.then([w = WeakFromThis(), read_stream, current_stream](auto f) {
+    if (auto self = w.lock()) {
+      self->OnFinish(read_stream, current_stream, f.get());
+    }
   });
 }
 
-void ObjectDescriptorImpl::OnFinish(StreamIterator it, Status const& status) {
+void ObjectDescriptorImpl::OnFinish(
+    std::shared_ptr<ReadStream> const& read_stream,
+    std::shared_ptr<OpenStream> const& stream, Status const& status) {
+  {
+    std::unique_lock<std::mutex> lk(mu_);
+    // Discard callbacks if cancelled or from stale/removed streams.
+    if (cancelled_) return;
+    auto it = stream_manager_->Find(read_stream);
+    if (it == stream_manager_->End() || !it->stream ||
+        it->stream->stream != stream) {
+      return;
+    }
+  }
   auto proto_status = ExtractGrpcStatus(status);
 
-  if (IsResumable(it, status, proto_status)) return Resume(it, proto_status);
+  if (IsResumable(read_stream, status, proto_status)) {
+    return Resume(read_stream, proto_status);
+  }
   std::unique_lock<std::mutex> lk(mu_);
+  // Re-verify stream identity under lock because IsResumable() releases and
+  // re-acquires the mutex while notifying range callbacks, during which time
+  // another thread or callback could have modified or replaced the stream.
+  if (cancelled_) return;
+  auto it = stream_manager_->Find(read_stream);
+  if (it == stream_manager_->End() || !it->stream ||
+      it->stream->stream != stream) {
+    return;
+  }
   stream_manager_->RemoveStreamAndNotifyRanges(it, status);
   // Since a stream died, we might want to ensure a replacement is queued.
   AssurePendingStreamQueued(lk);
 }
 
-void ObjectDescriptorImpl::Resume(StreamIterator it,
-                                  google::rpc::Status const& proto_status) {
+void ObjectDescriptorImpl::Resume(
+    std::shared_ptr<ReadStream> const& read_stream,
+    google::rpc::Status const& proto_status) {
   std::unique_lock<std::mutex> lk(mu_);
+  if (cancelled_) return;
+  auto it = stream_manager_->Find(read_stream);
+  if (it == stream_manager_->End() || !it->stream) return;
+  // Set resuming flag to true to prevent any concurrent Flush() from writing
+  // to the dying stream while we establish a new one.
+  it->stream->resuming = true;
+  it->stream->next_request.Clear();
+  auto current_stream = it->stream->stream;
   // This call needs to happen inside the lock, as it may modify
   // `read_object_spec_`.
   ApplyRedirectErrors(read_object_spec_, proto_status);
@@ -570,31 +664,71 @@ void ObjectDescriptorImpl::Resume(StreamIterator it,
     *request.add_read_ranges() = *std::move(range);
   }
   lk.unlock();
-  make_stream_(std::move(request)).then([w = WeakFromThis(), it](auto f) {
-    if (auto self = w.lock()) self->OnResume(it, f.get());
-  });
+  make_stream_(std::move(request))
+      .then([w = WeakFromThis(), read_stream, current_stream](auto f) {
+        if (auto self = w.lock()) {
+          self->OnResume(read_stream, current_stream, f.get());
+        }
+      });
 }
 
-void ObjectDescriptorImpl::OnResume(StreamIterator it,
-                                    StatusOr<OpenStreamResult> result) {
-  if (!result) return OnFinish(it, std::move(result).status());
+void ObjectDescriptorImpl::OnResume(
+    std::shared_ptr<ReadStream> const& old_read_stream,
+    std::shared_ptr<OpenStream> const& old_stream,
+    StatusOr<OpenStreamResult> result) {
+  {
+    std::unique_lock<std::mutex> lk(mu_);
+    if (cancelled_) {
+      if (result && result->stream) result->stream->Cancel();
+      return;
+    }
+  }
+  if (!result) {
+    return OnFinish(old_read_stream, old_stream, std::move(result).status());
+  }
   std::unique_lock<std::mutex> lk(mu_);
-  if (cancelled_) return;
+  // Discard resume responses if cancelled or if the stream entry was removed or
+  // already replaced.
+  auto it = stream_manager_->Find(old_read_stream);
+  if (cancelled_ || it == stream_manager_->End() || !it->stream ||
+      it->stream->stream != old_stream) {
+    if (result->stream) result->stream->Cancel();
+    return;
+  }
 
-  it->stream = std::make_shared<ReadStream>(std::move(result->stream),
-                                            resume_policy_prototype_->clone());
-  it->stream->write_pending = false;
-  it->stream->read_pending = false;
+  // Preserve any Read() range requests that arrived concurrently while the
+  // reconnection was in flight.
+  google::storage::v2::BidiReadObjectRequest queued_request =
+      std::move(it->stream->next_request);
+
+  // Replace the old stream with the new stream and preserve the existing
+  // resume policy so failure budgets and error counts are maintained across
+  // reconnects.
+  std::shared_ptr<ReadStream> new_read_stream = std::make_shared<ReadStream>(
+      std::move(result->stream), std::move(it->stream->resume_policy));
+  new_read_stream->resume_policy->OnStartSuccess();
+  new_read_stream->write_pending = false;
+  new_read_stream->read_pending = false;
+  new_read_stream->resuming = false;
+  new_read_stream->next_request = std::move(queued_request);
+
+  it->stream = new_read_stream;
+  std::shared_ptr<OpenStream> new_stream = new_read_stream->stream;
 
   // TODO(#15105) - this should be done without release the lock.
-  Flush(std::move(lk), it);
-  OnRead(it, std::move(result->first_response));
+  // Flush any queued range requests onto the newly active stream.
+  Flush(std::move(lk), new_read_stream);
+  // Process the first response received during stream establishment.
+  OnRead(new_read_stream, new_stream, std::move(result->first_response));
 }
 
 bool ObjectDescriptorImpl::IsResumable(
-    StreamIterator it, Status const& status,
+    std::shared_ptr<ReadStream> const& read_stream, Status const& status,
     google::rpc::Status const& proto_status) {
   std::unique_lock<std::mutex> lk(mu_);
+  if (cancelled_) return false;
+  auto it = stream_manager_->Find(read_stream);
+  if (it == stream_manager_->End() || !it->stream) return false;
   for (auto const& any : proto_status.details()) {
     auto error = google::storage::v2::BidiReadObjectError{};
     if (!any.UnpackTo(&error)) continue;
@@ -614,9 +748,16 @@ bool ObjectDescriptorImpl::IsResumable(
       if (l != copy.end()) l->second->OnFinish(p.second);
     }
     lk.lock();
-    stream_manager_->CleanupDoneRanges(it);
+    if (cancelled_) return true;
+    auto it_curr = stream_manager_->Find(read_stream);
+    if (it_curr == stream_manager_->End() || !it_curr->stream) return true;
+    stream_manager_->CleanupDoneRanges(it_curr);
     return true;
   }
+  // Pass the original status directly to the resume policy without rewriting
+  // status codes (such as StatusCode::kCancelled). This allows custom resume
+  // policies (e.g., detecting stall cancellations) to observe the exact failure
+  // cause.
   return it->stream->resume_policy->OnFinish(status) ==
          storage::ResumePolicy::kContinue;
 }
diff --git a/google/cloud/storage/internal/async/object_descriptor_impl.h b/google/cloud/storage/internal/async/object_descriptor_impl.h
index d0cfa98487c05..238764d8ecdc7 100644
--- a/google/cloud/storage/internal/async/object_descriptor_impl.h
+++ b/google/cloud/storage/internal/async/object_descriptor_impl.h
@@ -53,6 +53,7 @@ struct ReadStream : public storage_internal::StreamBase {
   google::storage::v2::BidiReadObjectRequest next_request;
   bool write_pending = false;
   bool read_pending = false;
+  bool resuming = false;
 };
 
 class ObjectDescriptorImpl
@@ -100,18 +101,32 @@ class ObjectDescriptorImpl
   // invoked while holding `mu_`.
   void AssurePendingStreamQueued(std::unique_lock<std::mutex> const&);
 
-  void Flush(std::unique_lock<std::mutex> lk, StreamIterator it);
-  void OnWrite(StreamIterator it, bool ok);
-  void DoRead(std::unique_lock<std::mutex> lk, StreamIterator it);
+  void Flush(std::unique_lock<std::mutex> lk,
+             std::shared_ptr<ReadStream> const& read_stream);
+  void OnWrite(std::shared_ptr<ReadStream> const& read_stream,
+               std::shared_ptr<OpenStream> const& stream, bool ok);
+  void DoRead(std::unique_lock<std::mutex> lk,
+              std::shared_ptr<ReadStream> const& read_stream);
   void OnRead(
-      StreamIterator it,
+      std::shared_ptr<ReadStream> const& read_stream,
+      std::shared_ptr<OpenStream> const& stream,
       std::optional<google::storage::v2::BidiReadObjectResponse> response);
-  void DoFinish(std::unique_lock<std::mutex> lk, StreamIterator it);
-  void OnFinish(StreamIterator it, Status const& status);
-  void Resume(StreamIterator it, google::rpc::Status const& proto_status);
-  void OnResume(StreamIterator it, StatusOr<OpenStreamResult> result);
-  bool IsResumable(StreamIterator it, Status const& status,
+  void DoFinish(std::unique_lock<std::mutex> lk,
+                std::shared_ptr<ReadStream> const& read_stream,
+                std::shared_ptr<OpenStream> const& stream);
+  void OnFinish(std::shared_ptr<ReadStream> const& read_stream,
+                std::shared_ptr<OpenStream> const& stream,
+                Status const& status);
+  void Resume(std::shared_ptr<ReadStream> const& read_stream,
+              google::rpc::Status const& proto_status);
+  void OnResume(std::shared_ptr<ReadStream> const& old_read_stream,
+                std::shared_ptr<OpenStream> const& old_stream,
+                StatusOr<OpenStreamResult> result);
+  bool IsResumable(std::shared_ptr<ReadStream> const& read_stream,
+                   Status const& status,
                    google::rpc::Status const& proto_status);
+  bool ApplyPacingAndCheckEviction(std::int64_t id, std::size_t chunk_size,
+                                   StreamIterator it);
 
   std::shared_ptr<storage::internal::HashFunction> CreateHashFunction(
       bool is_full_read) const;
diff --git a/google/cloud/storage/internal/async/object_descriptor_impl_test.cc b/google/cloud/storage/internal/async/object_descriptor_impl_test.cc
index d8ac5abd97024..3d39bfaac9ea2 100644
--- a/google/cloud/storage/internal/async/object_descriptor_impl_test.cc
+++ b/google/cloud/storage/internal/async/object_descriptor_impl_test.cc
@@ -24,6 +24,7 @@
 #include "google/cloud/storage/internal/async/default_options.h"
 #include "google/cloud/storage/options.h"
 #include "google/cloud/storage/testing/canonical_errors.h"
+#include "google/cloud/storage/testing/mock_resume_policy.h"
 #include "google/cloud/storage/testing/mock_storage_stub.h"
 #include "google/cloud/testing_util/async_sequencer.h"
 #include "google/cloud/testing_util/is_proto_equal.h"
@@ -41,6 +42,7 @@ namespace storage_internal {
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
 namespace {
 
+using ::google::cloud::storage::testing::MockResumePolicy;
 using ::google::cloud::storage::testing::canonical_errors::PermanentError;
 using ::google::cloud::storage::testing::canonical_errors::TransientError;
 using ::google::cloud::testing_util::AsyncSequencer;
@@ -49,8 +51,12 @@ using ::google::cloud::testing_util::IsProtoEqual;
 using ::google::cloud::testing_util::StatusIs;
 using ::google::protobuf::TextFormat;
 using ::testing::_;
+using ::testing::AnyNumber;
 using ::testing::AtMost;
 using ::testing::ElementsAre;
+using ::testing::Eq;
+using ::testing::IsFalse;
+using ::testing::IsTrue;
 using ::testing::NotNull;
 using ::testing::Optional;
 using ::testing::ResultOf;
@@ -3144,6 +3150,755 @@ TEST(ObjectDescriptorImpl, DuplicateInitialRangesDeduplication) {
   next.first.set_value(true);
 }
 
+/// @test Verify that when a stream fails and triggers resumption, concurrent
+/// Read() requests queued while reconnecting are not dropped and are flushed to
+/// the newly connected stream.
+TEST(ObjectDescriptorImpl, ResumeRacesWithConcurrentRead) {
+  AsyncSequencer<bool> sequencer;
+  auto stream1 = std::make_unique<MockStream>();
+  EXPECT_CALL(*stream1, Read).WillRepeatedly([&sequencer]() {
+    return sequencer.PushBack("Read[1]").then(
+        [](auto) { return std::optional<Response>{}; });
+  });
+  EXPECT_CALL(*stream1, Write)
+      .WillRepeatedly([&sequencer](Request const&, grpc::WriteOptions) {
+        return sequencer.PushBack("Write[1]").then([](auto f) {
+          return f.get();
+        });
+      });
+  EXPECT_CALL(*stream1, Finish).WillOnce([&sequencer]() {
+    return sequencer.PushBack("Finish[1]").then([](auto f) {
+      if (f.get()) return TransientError();
+      return PermanentError();
+    });
+  });
+
+  auto stream2 = std::make_unique<MockStream>();
+  EXPECT_CALL(*stream2, Read)
+      .WillOnce([&sequencer]() {
+        return sequencer.PushBack("Read[2]").then(
+            [](auto) { return std::optional<Response>{}; });
+      })
+      .WillRepeatedly(
+          []() { return make_ready_future(std::optional<Response>{}); });
+  EXPECT_CALL(*stream2, Write)
+      .WillRepeatedly([&sequencer](Request const&, grpc::WriteOptions) {
+        return sequencer.PushBack("Write[2]").then([](auto f) {
+          return f.get();
+        });
+      });
+  EXPECT_CALL(*stream2, Finish).WillRepeatedly([]() {
+    return make_ready_future(Status{});
+  });
+
+  MockFactory factory;
+  EXPECT_CALL(factory, Call).WillRepeatedly([](Request const&) {
+    return make_ready_future(StatusOr<OpenStreamResult>(TransientError()));
+  });
+  EXPECT_CALL(
+      factory,
+      Call(ResultOf([](Request const& r) { return !r.read_ranges().empty(); },
+                    true)))
+      .WillOnce([&stream2](Request const&) {
+        auto response = Response{};
+        return make_ready_future(make_status_or(
+            OpenStreamResult{std::make_shared<OpenStream>(std::move(stream2)),
+                             std::move(response)}));
+      });
+
+  auto tested = MakeTested(storage::LimitedErrorCountResumePolicy(3)(),
+                           factory.AsStdFunction(),
+                           google::storage::v2::BidiReadObjectSpec{},
+                           std::make_shared<OpenStream>(std::move(stream1)));
+
+  tested->Start(Response{});
+
+  auto read1 = sequencer.PopFrontWithName();
+  EXPECT_EQ(read1.second, "Read[1]");
+
+  auto s1 = tested->Read({1000, 100});
+  ASSERT_THAT(s1, NotNull());
+  auto write1 = sequencer.PopFrontWithName();
+  EXPECT_EQ(write1.second, "Write[1]");
+  write1.first.set_value(true);
+
+  // Trigger stream 1 failure and Resume
+  read1.first.set_value(false);
+
+  auto next = sequencer.PopFrontWithName();
+  EXPECT_EQ(next.second, "Finish[1]");
+  next.first.set_value(true);
+
+  // Concurrently issue Read for range 2 while Resume is connecting stream 2
+  auto s2 = tested->Read({2000, 100});
+  ASSERT_THAT(s2, NotNull());
+
+  // Stream 2 should be read and receive the flushed Write[2] with range 2
+  auto read2 = sequencer.PopFrontWithName();
+  EXPECT_EQ(read2.second, "Read[2]");
+
+  auto write2 = sequencer.PopFrontWithName();
+  EXPECT_EQ(write2.second, "Write[2]");
+  write2.first.set_value(true);
+
+  tested.reset();
+  read2.first.set_value(false);
+}
+
+/// @test Verify that callbacks from discarded/stale stream instances are
+/// ignored and do not affect the newly active stream.
+TEST(ObjectDescriptorImpl, IgnoresCallbacksFromStaleStreams) {
+  AsyncSequencer<bool> sequencer;
+  auto stream1 = std::make_unique<MockStream>();
+  EXPECT_CALL(*stream1, Read).WillRepeatedly([&sequencer]() {
+    return sequencer.PushBack("Read[1]").then(
+        [](auto) { return std::optional<Response>{}; });
+  });
+  EXPECT_CALL(*stream1, Write)
+      .WillRepeatedly([&sequencer](Request const&, grpc::WriteOptions) {
+        return sequencer.PushBack("Write[1]").then([](auto f) {
+          return f.get();
+        });
+      });
+  EXPECT_CALL(*stream1, Finish).WillOnce([&sequencer]() {
+    return sequencer.PushBack("Finish[1]").then([](auto f) {
+      if (f.get()) return TransientError();
+      return PermanentError();
+    });
+  });
+
+  auto stream2 = std::make_unique<MockStream>();
+  EXPECT_CALL(*stream2, Read)
+      .WillOnce([&sequencer]() {
+        return sequencer.PushBack("Read[2]").then(
+            [](auto) { return std::optional<Response>{}; });
+      })
+      .WillRepeatedly(
+          []() { return make_ready_future(std::optional<Response>{}); });
+  EXPECT_CALL(*stream2, Write)
+      .WillRepeatedly([&sequencer](Request const&, grpc::WriteOptions) {
+        return sequencer.PushBack("Write[2]").then([](auto f) {
+          return f.get();
+        });
+      });
+  EXPECT_CALL(*stream2, Finish).WillRepeatedly([]() {
+    return make_ready_future(Status{});
+  });
+
+  MockFactory factory;
+  EXPECT_CALL(factory, Call).WillRepeatedly([](Request const&) {
+    return make_ready_future(StatusOr<OpenStreamResult>(TransientError()));
+  });
+  EXPECT_CALL(
+      factory,
+      Call(ResultOf([](Request const& r) { return !r.read_ranges().empty(); },
+                    true)))
+      .WillOnce([&stream2](Request const&) {
+        auto response = Response{};
+        return make_ready_future(make_status_or(
+            OpenStreamResult{std::make_shared<OpenStream>(std::move(stream2)),
+                             std::move(response)}));
+      });
+
+  auto tested = MakeTested(storage::LimitedErrorCountResumePolicy(3)(),
+                           factory.AsStdFunction(),
+                           google::storage::v2::BidiReadObjectSpec{},
+                           std::make_shared<OpenStream>(std::move(stream1)));
+
+  tested->Start(Response{});
+
+  auto read1 = sequencer.PopFrontWithName();
+  EXPECT_EQ(read1.second, "Read[1]");
+
+  auto s1 = tested->Read({1000, 100});
+  ASSERT_THAT(s1, NotNull());
+  auto write1 = sequencer.PopFrontWithName();
+  EXPECT_EQ(write1.second, "Write[1]");
+
+  // Stream 1 fails on read and resumes stream 2
+  read1.first.set_value(false);
+
+  auto next = sequencer.PopFrontWithName();
+  EXPECT_EQ(next.second, "Finish[1]");
+  next.first.set_value(true);
+
+  // Stream 2 is now active
+  auto read2 = sequencer.PopFrontWithName();
+  EXPECT_EQ(read2.second, "Read[2]");
+
+  // A stale Write[1] callback from stream 1 arrives late with false (error).
+  // It should NOT call Finish[2] on stream 2!
+  write1.first.set_value(false);
+
+  EXPECT_TRUE(sequencer.empty());
+
+  tested.reset();
+  read2.first.set_value(false);
+}
+
+/// @test Verify that when a stream is cancelled due to a stall watchdog timeout
+/// (StatusCode::kCancelled), it is treated as resumable by the resume policy
+/// and automatically resumed.
+TEST(ObjectDescriptorImpl, ResumeOnStallTimeout) {
+  AsyncSequencer<bool> sequencer;
+  auto stream1 = std::make_unique<MockStream>();
+  EXPECT_CALL(*stream1, Read).WillRepeatedly([&sequencer]() {
+    return sequencer.PushBack("Read[1]").then(
+        [](auto) { return std::optional<Response>{}; });
+  });
+  EXPECT_CALL(*stream1, Write)
+      .WillRepeatedly([&sequencer](Request const&, grpc::WriteOptions) {
+        return sequencer.PushBack("Write[1]").then([](auto f) {
+          return f.get();
+        });
+      });
+  EXPECT_CALL(*stream1, Finish).WillOnce([&sequencer]() {
+    return sequencer.PushBack("Finish[1]").then([](auto f) {
+      if (f.get()) return Status(StatusCode::kCancelled, "Stream stalled");
+      return PermanentError();
+    });
+  });
+
+  auto stream2 = std::make_unique<MockStream>();
+  EXPECT_CALL(*stream2, Read)
+      .WillOnce([&sequencer]() {
+        return sequencer.PushBack("Read[2]").then(
+            [](auto) { return std::optional<Response>{}; });
+      })
+      .WillRepeatedly(
+          []() { return make_ready_future(std::optional<Response>{}); });
+  EXPECT_CALL(*stream2, Write)
+      .WillRepeatedly([&sequencer](Request const&, grpc::WriteOptions) {
+        return sequencer.PushBack("Write[2]").then([](auto f) {
+          return f.get();
+        });
+      });
+  EXPECT_CALL(*stream2, Finish).WillRepeatedly([]() {
+    return make_ready_future(Status{});
+  });
+
+  MockFactory factory;
+  EXPECT_CALL(factory, Call).WillRepeatedly([](Request const&) {
+    return make_ready_future(StatusOr<OpenStreamResult>(TransientError()));
+  });
+  EXPECT_CALL(
+      factory,
+      Call(ResultOf([](Request const& r) { return !r.read_ranges().empty(); },
+                    true)))
+      .WillOnce([&stream2](Request const&) {
+        auto response = Response{};
+        return make_ready_future(make_status_or(
+            OpenStreamResult{std::make_shared<OpenStream>(std::move(stream2)),
+                             std::move(response)}));
+      });
+
+  auto tested = MakeTested(storage::LimitedErrorCountResumePolicy(3)(),
+                           factory.AsStdFunction(),
+                           google::storage::v2::BidiReadObjectSpec{},
+                           std::make_shared<OpenStream>(std::move(stream1)));
+
+  tested->Start(Response{});
+
+  auto read1 = sequencer.PopFrontWithName();
+  EXPECT_EQ(read1.second, "Read[1]");
+
+  auto s1 = tested->Read({1000, 100});
+  ASSERT_THAT(s1, NotNull());
+  auto write1 = sequencer.PopFrontWithName();
+  EXPECT_EQ(write1.second, "Write[1]");
+  write1.first.set_value(true);
+
+  // Watchdog cancels stream 1 on read stall
+  read1.first.set_value(false);
+
+  auto next = sequencer.PopFrontWithName();
+  EXPECT_EQ(next.second, "Finish[1]");
+  next.first.set_value(true);
+
+  // If resumed, stream 2 is connected and Read[2] is invoked
+  auto read2 = sequencer.PopFrontWithName();
+  EXPECT_EQ(read2.second, "Read[2]");
+
+  tested.reset();
+  read2.first.set_value(false);
+}
+
+/// @test Verify that when a stream starts successfully, the resume policy is
+/// notified via OnStartSuccess().
+TEST(ObjectDescriptorImpl, NotifiesResumePolicyOnStartSuccess) {
+  auto mock_policy = std::make_unique<MockResumePolicy>();
+  EXPECT_CALL(*mock_policy, clone).WillOnce([]() {
+    auto p = std::make_unique<MockResumePolicy>();
+    EXPECT_CALL(*p, OnStartSuccess).Times(::testing::AtLeast(1));
+    return p;
+  });
+
+  MockFactory factory;
+  EXPECT_CALL(factory, Call).WillRepeatedly([](Request const&) {
+    return make_ready_future(StatusOr<OpenStreamResult>(TransientError()));
+  });
+  auto stream = std::make_unique<MockStream>();
+  EXPECT_CALL(*stream, Read).WillRepeatedly([]() {
+    return make_ready_future(std::optional<Response>{});
+  });
+  EXPECT_CALL(*stream, Write)
+      .WillRepeatedly([](Request const&, grpc::WriteOptions) {
+        return make_ready_future(true);
+      });
+  EXPECT_CALL(*stream, Finish).WillRepeatedly([]() {
+    return make_ready_future(Status{});
+  });
+
+  auto tested = MakeTested(std::move(mock_policy), factory.AsStdFunction(),
+                           google::storage::v2::BidiReadObjectSpec{},
+                           std::make_shared<OpenStream>(std::move(stream)));
+
+  tested->Start(Response{});
+}
+
+/// @test Verify that when the user explicitly cancels ObjectDescriptorImpl,
+/// streams are cancelled and no resume attempts are made.
+TEST(ObjectDescriptorImpl, UserCancelStopsStreamWithoutResume) {
+  AsyncSequencer<bool> sequencer;
+  auto stream1 = std::make_unique<MockStream>();
+  EXPECT_CALL(*stream1, Cancel).Times(::testing::AtLeast(1));
+  EXPECT_CALL(*stream1, Write)
+      .Times(AtMost(1))
+      .WillRepeatedly([](Request const&, grpc::WriteOptions) {
+        return make_ready_future(true);
+      });
+  EXPECT_CALL(*stream1, Read).WillOnce([&sequencer]() {
+    return sequencer.PushBack("Read[1]").then(
+        [](auto) { return std::optional<Response>{}; });
+  });
+  EXPECT_CALL(*stream1, Finish).WillOnce([&sequencer]() {
+    return sequencer.PushBack("Finish[1]").then([](auto) {
+      return Status(StatusCode::kCancelled, "Cancelled by user");
+    });
+  });
+
+  MockFactory factory;
+  // Factory should NEVER be called on user cancellation.
+  EXPECT_CALL(factory, Call).Times(0);
+
+  Options options;
+  options.set<storage::EnableMultiStreamOptimizationOption>(false);
+  auto tested = std::make_shared<ObjectDescriptorImpl>(
+      storage::LimitedErrorCountResumePolicy(2)(), factory.AsStdFunction(),
+      google::storage::v2::BidiReadObjectSpec{},
+      std::make_shared<OpenStream>(std::move(stream1)), options);
+
+  Response start_resp;
+  tested->Start(std::move(start_resp));
+
+  auto r1 = tested->Read({100, 50});
+  ASSERT_THAT(r1, NotNull());
+
+  auto read1 = sequencer.PopFrontWithName();
+  EXPECT_EQ(read1.second, "Read[1]");
+
+  // User explicitly cancels.
+  tested->Cancel();
+  EXPECT_FALSE(tested->IsOpen());
+
+  read1.first.set_value(true);
+
+  auto finish1 = sequencer.PopFrontWithName();
+  EXPECT_EQ(finish1.second, "Finish[1]");
+  finish1.first.set_value(true);
+
+  // No resume factory calls happen.
+  EXPECT_TRUE(sequencer.empty());
+  EXPECT_FALSE(tested->IsOpen());
+}
+
+/// @test Verify that range requests arriving during multiple consecutive
+/// transient reconnection retries are preserved across all attempts and
+/// flushed to the new stream once reconnected.
+TEST(ObjectDescriptorImpl, ConcurrentReadPreservedAcrossMultipleResumeRetries) {
+  AsyncSequencer<bool> sequencer;
+  auto stream1 = std::make_unique<MockStream>();
+  EXPECT_CALL(*stream1, Read).WillRepeatedly([&sequencer]() {
+    return sequencer.PushBack("Read[1]").then(
+        [](auto) { return std::optional<Response>{}; });
+  });
+  EXPECT_CALL(*stream1, Write)
+      .WillRepeatedly([&sequencer](Request const&, grpc::WriteOptions) {
+        return sequencer.PushBack("Write[1]").then([](auto f) {
+          return f.get();
+        });
+      });
+  EXPECT_CALL(*stream1, Finish).WillOnce([&sequencer]() {
+    return sequencer.PushBack("Finish[1]").then([](auto f) {
+      if (f.get()) return TransientError();
+      return PermanentError();
+    });
+  });
+
+  auto stream2 = std::make_unique<MockStream>();
+  EXPECT_CALL(*stream2, Read)
+      .WillOnce([&sequencer]() {
+        return sequencer.PushBack("Read[2]").then(
+            [](auto) { return std::optional<Response>{}; });
+      })
+      .WillRepeatedly(
+          []() { return make_ready_future(std::optional<Response>{}); });
+  EXPECT_CALL(*stream2, Write)
+      .WillRepeatedly([&sequencer](Request const&, grpc::WriteOptions) {
+        return sequencer.PushBack("Write[2]").then([](auto f) {
+          return f.get();
+        });
+      });
+  EXPECT_CALL(*stream2, Finish).WillRepeatedly([]() {
+    return make_ready_future(Status{});
+  });
+
+  MockFactory factory;
+  int retry_count = 0;
+  EXPECT_CALL(factory, Call).WillRepeatedly([&](Request const& r) {
+    ++retry_count;
+    if (retry_count == 1) {
+      return sequencer.PushBack("Factory[1]").then([](auto f) {
+        if (f.get()) {
+          return StatusOr<OpenStreamResult>(TransientError());
+        }
+        return StatusOr<OpenStreamResult>(PermanentError());
+      });
+    }
+    // Second attempt must contain both range 1 and range 2
+    EXPECT_EQ(r.read_ranges_size(), 2);
+    return sequencer.PushBack("Factory[2]").then([&stream2](auto f) {
+      if (f.get()) {
+        auto response = Response{};
+        return make_status_or(
+            OpenStreamResult{std::make_shared<OpenStream>(std::move(stream2)),
+                             std::move(response)});
+      }
+      return StatusOr<OpenStreamResult>(PermanentError());
+    });
+  });
+
+  auto tested = std::make_shared<ObjectDescriptorImpl>(
+      storage::LimitedErrorCountResumePolicy(3)(), factory.AsStdFunction(),
+      google::storage::v2::BidiReadObjectSpec{},
+      std::make_shared<OpenStream>(std::move(stream1)),
+      Options{}.set<storage::EnableMultiStreamOptimizationOption>(false));
+
+  tested->Start(Response{});
+
+  auto read1 = sequencer.PopFrontWithName();
+  EXPECT_EQ(read1.second, "Read[1]");
+
+  auto s1 = tested->Read({1000, 100});
+  ASSERT_THAT(s1, NotNull());
+  auto write1 = sequencer.PopFrontWithName();
+  EXPECT_EQ(write1.second, "Write[1]");
+  write1.first.set_value(true);
+
+  // Trigger stream 1 failure
+  read1.first.set_value(false);
+
+  auto finish1 = sequencer.PopFrontWithName();
+  EXPECT_EQ(finish1.second, "Finish[1]");
+  finish1.first.set_value(true);
+
+  // Factory attempt 1 is invoked
+  auto factory1 = sequencer.PopFrontWithName();
+  EXPECT_EQ(factory1.second, "Factory[1]");
+
+  // While reconnection attempt 1 is in-flight, add range 2 concurrently
+  auto s2 = tested->Read({2000, 100});
+  ASSERT_THAT(s2, NotNull());
+
+  // Factory attempt 1 fails with TransientError
+  factory1.first.set_value(true);
+
+  // Factory attempt 2 is invoked with both range 1 and range 2
+  auto factory2 = sequencer.PopFrontWithName();
+  EXPECT_EQ(factory2.second, "Factory[2]");
+
+  // While factory attempt 2 is in-flight, add range 3 concurrently
+  auto s3 = tested->Read({3000, 100});
+  ASSERT_THAT(s3, NotNull());
+
+  // Factory attempt 2 succeeds with stream2
+  factory2.first.set_value(true);
+
+  // Range 3 (queued during factory attempt 2) is flushed to stream 2 first
+  auto write2 = sequencer.PopFrontWithName();
+  EXPECT_EQ(write2.second, "Write[2]");
+  write2.first.set_value(true);
+
+  // Stream 2 is read
+  auto read2 = sequencer.PopFrontWithName();
+  EXPECT_EQ(read2.second, "Read[2]");
+
+  EXPECT_EQ(retry_count, 2);
+
+  tested->Cancel();
+  read2.first.set_value(false);
+}
+
+/// @test Verify that a stream waiting for a reconnection is not offered as an
+/// "idle" stream, and that new reads are not parked on it.
+TEST(ObjectDescriptorImpl, ResumingStreamIsNotReusedAsIdleStream) {
+  AsyncSequencer<bool> sequencer;
+  AsyncSequencer<bool> factory_sequencer;
+  auto stream1 = std::make_unique<MockStream>();
+  EXPECT_CALL(*stream1, Read).WillOnce([&sequencer]() {
+    return sequencer.PushBack("Read[1]").then(
+        [](auto) { return std::optional<Response>{}; });
+  });
+  EXPECT_CALL(*stream1, Finish).WillOnce([&sequencer]() {
+    return sequencer.PushBack("Finish[1]").then([](auto) {
+      return TransientError();
+    });
+  });
+  EXPECT_CALL(*stream1, Cancel).Times(AtMost(1));
+
+  auto stream2 = std::make_unique<MockStream>();
+  EXPECT_CALL(*stream2, Read).WillRepeatedly([&sequencer]() {
+    return sequencer.PushBack("Read[2]").then(
+        [](auto) { return std::optional<Response>{}; });
+  });
+  // The read issued while stream 1 is reconnecting must be written to the
+  // healthy stream.
+  EXPECT_CALL(*stream2, Write)
+      .WillOnce([&sequencer](Request const&, grpc::WriteOptions) {
+        return sequencer.PushBack("Write[2]").then([](auto f) {
+          return f.get();
+        });
+      });
+  EXPECT_CALL(*stream2, Finish).Times(AtMost(1)).WillRepeatedly([&sequencer]() {
+    return sequencer.PushBack("Finish[2]").then([](auto) { return Status{}; });
+  });
+  EXPECT_CALL(*stream2, Cancel).Times(AtMost(1));
+
+  // The first factory call is the proactive background stream queued by
+  // `Start()`; it produces stream 2. Any later call (including the
+  // reconnection attempt for stream 1) fails, and the reconnection is left
+  // in flight for most of the test.
+  int factory_calls = 0;
+  MockFactory factory;
+  EXPECT_CALL(factory, Call).WillRepeatedly([&](Request const&) {
+    int const call = ++factory_calls;
+    char const* name = call == 1 ? "Factory[background]" : "Factory[resume]";
+    return factory_sequencer.PushBack(name).then(
+        [&stream2, call](auto f) -> StatusOr<OpenStreamResult> {
+          if (call != 1 || !f.get()) return TransientError();
+          return make_status_or(OpenStreamResult{
+              std::make_shared<OpenStream>(std::move(stream2)), Response{}});
+        });
+  });
+
+  auto tested = MakeTested(storage::LimitedErrorCountResumePolicy(3)(),
+                           factory.AsStdFunction(),
+                           google::storage::v2::BidiReadObjectSpec{},
+                           std::make_shared<OpenStream>(std::move(stream1)));
+
+  tested->Start(Response{});
+
+  auto read1 = sequencer.PopFrontWithName();
+  EXPECT_THAT(read1.second, Eq("Read[1]"));
+  auto background = factory_sequencer.PopFrontWithName();
+  EXPECT_THAT(background.second, Eq("Factory[background]"));
+
+  // Stream 1 fails while it has no active ranges, e.g. the stall watchdog
+  // cancelled an idle stream. The descriptor starts a reconnection which does
+  // not complete yet, leaving the entry in the `resuming` state.
+  read1.first.set_value(false);
+  auto finish1 = sequencer.PopFrontWithName();
+  EXPECT_THAT(finish1.second, Eq("Finish[1]"));
+  finish1.first.set_value(true);
+  auto reconnect = factory_sequencer.PopFrontWithName();
+  EXPECT_THAT(reconnect.second, Eq("Factory[resume]"));
+
+  // The application asks for an additional stream. The only entry is waiting
+  // to reconnect, so the descriptor must consume the background stream
+  // instead of reporting that it reused an idle one.
+  tested->MakeSubsequentStream();
+  background.first.set_value(true);
+  EXPECT_THAT(tested->StreamSize(), Eq(std::size_t{2}));
+
+  // The new read must go to the healthy stream. If it is queued on the
+  // reconnecting stream `Flush()` is skipped and nothing reaches the wire.
+  std::unique_ptr<storage::AsyncReaderConnection> reader =
+      tested->Read({0, 100});
+  EXPECT_THAT(reader, NotNull());
+  EXPECT_THAT(sequencer.empty(), IsFalse());
+
+  tested.reset();
+  reconnect.first.set_value(false);
+  while (!sequencer.empty()) sequencer.PopFront().set_value(false);
+  while (!factory_sequencer.empty()) {
+    factory_sequencer.PopFront().set_value(false);
+  }
+}
+
+/// @test Verify the resume policy is told the actual status of the stream.
+TEST(ObjectDescriptorImpl, ResumePolicyObservesCancelledStatus) {
+  AsyncSequencer<bool> sequencer;
+  std::optional<Status> policy_status;
+  auto prototype = std::make_unique<MockResumePolicy>();
+  EXPECT_CALL(*prototype, clone).WillRepeatedly([&policy_status]() {
+    auto policy = std::make_unique<MockResumePolicy>();
+    EXPECT_CALL(*policy, OnStartSuccess).Times(AnyNumber());
+    EXPECT_CALL(*policy, OnFinish)
+        .WillRepeatedly([&policy_status](Status const& status) {
+          policy_status = status;
+          return storage::ResumePolicy::kStop;
+        });
+    return policy;
+  });
+
+  auto stream1 = std::make_unique<MockStream>();
+  EXPECT_CALL(*stream1, Read).WillOnce([&sequencer]() {
+    return sequencer.PushBack("Read[1]").then(
+        [](auto) { return std::optional<Response>{}; });
+  });
+  EXPECT_CALL(*stream1, Finish).WillOnce([&sequencer]() {
+    return sequencer.PushBack("Finish[1]").then([](auto) {
+      return Status(StatusCode::kCancelled, "Stream stalled");
+    });
+  });
+  EXPECT_CALL(*stream1, Cancel).Times(AtMost(1));
+
+  MockFactory factory;
+  EXPECT_CALL(factory, Call).WillRepeatedly([](Request const&) {
+    return make_ready_future(StatusOr<OpenStreamResult>(PermanentError()));
+  });
+
+  Options options;
+  options.set<storage::EnableMultiStreamOptimizationOption>(false);
+  auto tested = std::make_shared<ObjectDescriptorImpl>(
+      std::move(prototype), factory.AsStdFunction(),
+      google::storage::v2::BidiReadObjectSpec{},
+      std::make_shared<OpenStream>(std::move(stream1)), options);
+
+  tested->Start(Response{});
+
+  auto read1 = sequencer.PopFrontWithName();
+  EXPECT_THAT(read1.second, Eq("Read[1]"));
+  read1.first.set_value(false);
+
+  auto finish1 = sequencer.PopFrontWithName();
+  EXPECT_THAT(finish1.second, Eq("Finish[1]"));
+  finish1.first.set_value(true);
+
+  EXPECT_THAT(policy_status.has_value(), IsTrue());
+  if (policy_status.has_value()) {
+    EXPECT_THAT(*policy_status,
+                StatusIs(StatusCode::kCancelled, "Stream stalled"));
+  }
+
+  tested.reset();
+  while (!sequencer.empty()) sequencer.PopFront().set_value(false);
+}
+
+/// @test Verify the resume budget is consumed across successful reconnects.
+TEST(ObjectDescriptorImpl, ResumeBudgetIsNotResetByASuccessfulReconnect) {
+  AsyncSequencer<bool> sequencer;
+  auto stream1 = std::make_unique<MockStream>();
+  EXPECT_CALL(*stream1, Read).WillOnce([&sequencer]() {
+    return sequencer.PushBack("Read[1]").then(
+        [](auto) { return std::optional<Response>{}; });
+  });
+  EXPECT_CALL(*stream1, Write)
+      .WillOnce([&sequencer](Request const&, grpc::WriteOptions) {
+        return sequencer.PushBack("Write[1]").then([](auto f) {
+          return f.get();
+        });
+      });
+  EXPECT_CALL(*stream1, Finish).WillOnce([&sequencer]() {
+    return sequencer.PushBack("Finish[1]").then([](auto) {
+      return TransientError();
+    });
+  });
+  EXPECT_CALL(*stream1, Cancel).Times(AtMost(1));
+
+  auto stream2 = std::make_unique<MockStream>();
+  EXPECT_CALL(*stream2, Read).WillOnce([&sequencer]() {
+    return sequencer.PushBack("Read[2]").then(
+        [](auto) { return std::optional<Response>{}; });
+  });
+  EXPECT_CALL(*stream2, Write)
+      .Times(AtMost(1))
+      .WillRepeatedly([&sequencer](Request const&, grpc::WriteOptions) {
+        return sequencer.PushBack("Write[2]").then([](auto f) {
+          return f.get();
+        });
+      });
+  EXPECT_CALL(*stream2, Finish).WillOnce([&sequencer]() {
+    return sequencer.PushBack("Finish[2]").then([](auto) {
+      return TransientError();
+    });
+  });
+  EXPECT_CALL(*stream2, Cancel).Times(AtMost(1));
+
+  // Reconnection attempts carry the ranges to resume; the proactive background
+  // streams do not. Only the former are counted.
+  int resume_attempts = 0;
+  MockFactory factory;
+  EXPECT_CALL(factory, Call)
+      .WillRepeatedly(
+          [&](Request const& request) -> future<StatusOr<OpenStreamResult>> {
+            if (request.read_ranges().empty()) {
+              return make_ready_future(
+                  StatusOr<OpenStreamResult>(TransientError()));
+            }
+            if (++resume_attempts != 1) {
+              return make_ready_future(
+                  StatusOr<OpenStreamResult>(TransientError()));
+            }
+            return make_ready_future(make_status_or(OpenStreamResult{
+                std::make_shared<OpenStream>(std::move(stream2)), Response{}}));
+          });
+
+  Options options;
+  options.set<storage::EnableMultiStreamOptimizationOption>(false);
+  // A budget of exactly one resume for the lifetime of this descriptor.
+  auto tested = std::make_shared<ObjectDescriptorImpl>(
+      storage::LimitedErrorCountResumePolicy(1)(), factory.AsStdFunction(),
+      google::storage::v2::BidiReadObjectSpec{},
+      std::make_shared<OpenStream>(std::move(stream1)), options);
+
+  tested->Start(Response{});
+
+  // `Start()` leaves a read outstanding on stream 1.
+  auto read1 = sequencer.PopFrontWithName();
+  EXPECT_THAT(read1.second, Eq("Read[1]"));
+
+  std::unique_ptr<storage::AsyncReaderConnection> reader =
+      tested->Read({0, 100});
+  EXPECT_THAT(reader, NotNull());
+  auto write1 = sequencer.PopFrontWithName();
+  EXPECT_THAT(write1.second, Eq("Write[1]"));
+  write1.first.set_value(true);
+
+  // First failure: consumes the single resume allowed by the policy.
+  read1.first.set_value(false);
+  auto finish1 = sequencer.PopFrontWithName();
+  EXPECT_THAT(finish1.second, Eq("Finish[1]"));
+  finish1.first.set_value(true);
+  EXPECT_THAT(resume_attempts, Eq(1));
+
+  // Second failure, on the replacement stream. The budget is exhausted, so the
+  // descriptor must give up instead of reconnecting again.
+  auto read2 = sequencer.PopFrontWithName();
+  EXPECT_THAT(read2.second, Eq("Read[2]"));
+  read2.first.set_value(false);
+
+  auto finish2 = sequencer.PopFrontWithName();
+  EXPECT_THAT(finish2.second, Eq("Finish[2]"));
+  finish2.first.set_value(true);
+
+  EXPECT_THAT(resume_attempts, Eq(1));
+  EXPECT_THAT(tested->IsOpen(), IsFalse());
+
+  tested.reset();
+  while (!sequencer.empty()) sequencer.PopFront().set_value(false);
+}
+
 }  // namespace
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
 }  // namespace storage_internal