diff --git a/doc/design/storage_download_api_design.md b/doc/design/storage_download_api_design.md
index 29133018d..0cb738322 100644
--- a/doc/design/storage_download_api_design.md
+++ b/doc/design/storage_download_api_design.md
@@ -20,17 +20,20 @@ func readObject(
   from bucket: String,
   object: String,
   options: ReadObjectOptions = .init()
-) async throws -> ReadObjectResult
+) -> ReadObjectTask
 ```
 
-- **Return Container Struct (`ReadObjectResult`):** `readObject` is an `async throws` function that performs the initial HTTP handshake and returns a container struct holding both the object metadata and the streaming body:
+- **Return Task Struct (`ReadObjectTask`):** `readObject` returns immediately with a `ReadObjectTask` struct holding both the object metadata and the streaming body:
   ```swift
-  public struct ReadObjectResult: Sendable {
+  public struct ReadObjectTask: Sendable {
     /// Object metadata populated from response headers upon request initiation.
-    public let metadata: ReadObjectMetadata
+    public var metadata: ReadObjectMetadata { get async throws }
 
-    /// An asynchronous sequence of `Data` chunks for the object payload.
-    public let body: ReadObjectSequence
+    /// An asynchronous sequence of `ByteBuffer` chunks for the object payload.
+    public var body: ReadObjectSequence { get }
+
+    /// Cancels the ongoing download.
+    public func cancel()
   }
   ```
 
@@ -297,12 +300,15 @@ public struct ReadObjectSequence: AsyncSequence, Sendable {
 }
 
 /// Container object returned by `readObject` containing metadata and the streaming body sequence.
-public struct ReadObjectResult: Sendable {
+public struct ReadObjectTask: Sendable {
   /// Object metadata extracted from initial HTTP response headers.
-  public let metadata: ReadObjectMetadata
+  public var metadata: ReadObjectMetadata { get async throws }
 
   /// Asynchronous sequence yielding chunks of binary data payload.
-  public let body: ReadObjectSequence
+  public var body: ReadObjectSequence { get }
+
+  /// Cancels the ongoing download.
+  public func cancel()
 }
 ```
 
@@ -314,15 +320,15 @@ public protocol StorageClientProtocol {
     from bucket: String,
     object: String,
     options: ReadObjectOptions
-  ) async throws -> ReadObjectResult
+  ) -> ReadObjectTask
 }
 
 extension StorageClientProtocol {
   public func readObject(
     from bucket: String,
     object: String
-  ) async throws -> ReadObjectResult {
-    try await readObject(from: bucket, object: object, options: .init())
+  ) -> ReadObjectTask {
+    readObject(from: bucket, object: object, options: .init())
   }
 }
 
@@ -331,10 +337,8 @@ extension StorageClient {
     from bucket: String,
     object: String,
     options: ReadObjectOptions = .init()
-  ) async throws -> ReadObjectResult {
-    // 1. Send initial GET request header handshake
-    // 2. Parse response headers into ReadObjectMetadata
-    // 3. Construct and return ReadObjectResult(metadata: metadata, body: sequence)
+  ) -> ReadObjectTask {
+    // Return ReadObjectTask backed by coordinator
   }
 }
 ```
diff --git a/packages/storage/Sources/GoogleCloudStorage/DownloadOptions.swift b/packages/storage/Sources/GoogleCloudStorage/DownloadOptions.swift
index 2d8cebc57..b85078129 100644
--- a/packages/storage/Sources/GoogleCloudStorage/DownloadOptions.swift
+++ b/packages/storage/Sources/GoogleCloudStorage/DownloadOptions.swift
@@ -176,6 +176,12 @@ public struct ReadObjectOptions: Sendable {
   /// Flag to enable transparent auto-resumption on transient network failures. Defaults to `true`.
   public var autoResume: Bool = true
 
+  /// Overrides the retry policy for this download.
+  public var retryPolicy: (any RetryPolicy)? = nil
+
+  /// Overrides the backoff policy for this download.
+  public var backoffPolicy: (any BackoffPolicy)? = nil
+
   /// Default configuration options.
   public static var `default`: ReadObjectOptions { ReadObjectOptions() }
 
@@ -190,6 +196,41 @@ public struct ReadObjectOptions: Sendable {
   }
 }
 
+/// Calculates the remaining range to request when resuming an interrupted download.
+///
+/// - Parameters:
+///   - originalRange: The range requested in the original download operation.
+///   - bytesReceived: The number of bytes successfully received and yielded so far.
+///   - totalSize: The total size of the object if known from metadata or headers.
+/// - Returns: The adjusted `ReadObjectRange` to request, or `nil` if all requested bytes have been received.
+package func calculateResumeRange(
+  originalRange: ReadObjectRange,
+  bytesReceived: UInt64,
+  totalSize: UInt64?
+) -> ReadObjectRange? {
+  switch originalRange {
+  case .entire:
+    return .fromOffset(bytesReceived)
+  case .fromOffset(let offset):
+    return .fromOffset(offset + bytesReceived)
+  case .prefix(let count):
+    guard count > bytesReceived else { return nil }
+    return .bounded(start: bytesReceived, end: count - 1)
+  case .bounded(let start, let end):
+    let newStart = start + bytesReceived
+    guard newStart <= end else { return nil }
+    return .bounded(start: newStart, end: end)
+  case .suffix(let count):
+    guard let totalSize = totalSize, totalSize > 0 else {
+      return .fromOffset(bytesReceived)
+    }
+    let startOffset = totalSize > count ? (totalSize - count) : 0
+    let newStart = startOffset + bytesReceived
+    guard newStart < totalSize else { return nil }
+    return .bounded(start: newStart, end: totalSize - 1)
+  }
+}
+
 /// Metadata attributes for an object returned in response headers during a download.
 public struct ReadObjectMetadata: Sendable, Hashable, Equatable {
   /// Name of the bucket containing the object.
@@ -246,122 +287,298 @@ public struct ReadObjectMetadata: Sendable, Hashable, Equatable {
 public struct ReadObjectSequence: AsyncSequence, Sendable {
   public typealias Element = NIOCore.ByteBuffer
 
-  /// Name of the bucket containing the object being read.
-  public var bucket: String = ""
+  private let coordinator: ReadObjectCoordinator
 
-  /// Name of the object being read.
-  public var object: String = ""
+  package init(coordinator: ReadObjectCoordinator) {
+    self.coordinator = coordinator
+  }
 
-  /// Configuration options used for this object download.
-  public var options: ReadObjectOptions = .init()
+  /// An asynchronous iterator for iterating over chunks of downloaded object payload data.
+  public struct AsyncIterator: AsyncIteratorProtocol {
+    public typealias Element = NIOCore.ByteBuffer
 
-  /// Object metadata extracted from initial HTTP response headers.
-  public var metadata: ReadObjectMetadata = .init()
+    private let coordinator: ReadObjectCoordinator
 
-  package var initialBody: _HTTPResponseBody?
-  package var stream: AsyncThrowingStream<NIOCore.ByteBuffer, Error>?
+    package init(coordinator: ReadObjectCoordinator) {
+      self.coordinator = coordinator
+    }
 
-  /// Creates a new `ReadObjectSequence` instance.
-  public init() {}
+    /// Advances to the next `ByteBuffer` chunk in the downloaded object payload stream.
+    public mutating func next() async throws -> NIOCore.ByteBuffer? {
+      try await coordinator.nextChunk()
+    }
+  }
 
-  /// Builder pattern helper to modify configuration in place.
-  public func with(_ config: (inout Self) -> Void) -> Self {
-    var copy = self
-    config(&copy)
-    return copy
+  /// Creates an asynchronous iterator for iterating over object payload chunks.
+  public func makeAsyncIterator() -> AsyncIterator {
+    AsyncIterator(coordinator: coordinator)
   }
+}
 
-  /// An asynchronous iterator for iterating over chunks of downloaded object payload data.
-  public struct AsyncIterator: AsyncIteratorProtocol, Sendable {
-    public typealias Element = NIOCore.ByteBuffer
+/// Coordinates the deferred initial request, metadata resolution, and streaming body consumption.
+package final class ReadObjectCoordinator: @unchecked Sendable {
+  let bucket: String
+  let object: String
+  let options: ReadObjectOptions
+  let httpClient: GoogleCloudGax._HTTPClient
+  let retryLoop: _RetryLoop
+
+  private let lock = NSLock()
+  private var isInitialFetched: Bool = false
+  private var initialFetchTask: Task<ReadObjectMetadata, Error>?
+  private var metadata: ReadObjectMetadata?
+  private var bodyIterator: _HTTPResponseBody.AsyncIterator?
+  private var streamIterator: AsyncThrowingStream<NIOCore.ByteBuffer, Error>.AsyncIterator?
+  private var bytesReceived: UInt64 = 0
+  private var isFinished: Bool = false
+  private var isCancelled: Bool = false
+
+  package init(
+    bucket: String,
+    object: String,
+    options: ReadObjectOptions,
+    httpClient: GoogleCloudGax._HTTPClient,
+    retryLoop: _RetryLoop
+  ) {
+    self.bucket = bucket
+    self.object = object
+    self.options = options
+    self.httpClient = httpClient
+    self.retryLoop = retryLoop
+  }
 
-    package final class Storage: @unchecked Sendable {
-      let options: ReadObjectOptions
-      var bodyIterator: _HTTPResponseBody.AsyncIterator?
-      var streamIterator: AsyncThrowingStream<NIOCore.ByteBuffer, Error>.AsyncIterator?
-      var isFinished: Bool = false
-
-      init(
-        options: ReadObjectOptions,
-        initialBody: _HTTPResponseBody?,
-        stream: AsyncThrowingStream<NIOCore.ByteBuffer, Error>?
-      ) {
-        self.options = options
-        self.bodyIterator = initialBody?.makeAsyncIterator()
-        self.streamIterator = stream?.makeAsyncIterator()
+  private func ensureInitialFetch() async throws -> ReadObjectMetadata {
+    if isCancelled {
+      throw CancellationError()
+    }
+    let task = lock.withLock {
+      if let existingTask = initialFetchTask {
+        return existingTask
       }
+      let newTask = Task { () -> ReadObjectMetadata in
+        let (response, metadata) = try await Self.fetchInitial(
+          httpClient: self.httpClient,
+          bucket: self.bucket,
+          object: self.object,
+          options: self.options,
+          retryLoop: self.retryLoop
+        )
+        self.lock.withLock {
+          self.metadata = metadata
+          self.bodyIterator = response.body.makeAsyncIterator()
+          self.isInitialFetched = true
+        }
+        return metadata
+      }
+      self.initialFetchTask = newTask
+      return newTask
+    }
+    return try await task.value
+  }
 
-      func next() async throws -> NIOCore.ByteBuffer? {
-        guard !isFinished else { return nil }
+  package func getMetadata() async throws -> ReadObjectMetadata {
+    if isCancelled {
+      throw CancellationError()
+    }
+    return try await ensureInitialFetch()
+  }
 
-        if case .prefix(0) = options.range {
-          isFinished = true
-          return nil
-        }
-        if case .suffix(0) = options.range {
-          isFinished = true
-          return nil
-        }
+  package func nextChunk() async throws -> NIOCore.ByteBuffer? {
+    guard !isFinished && !isCancelled else { return nil }
+
+    if case .prefix(0) = options.range {
+      isFinished = true
+      return nil
+    }
+    if case .suffix(0) = options.range {
+      isFinished = true
+      return nil
+    }
 
+    _ = try await ensureInitialFetch()
+
+    while !isFinished && !isCancelled {
+      do {
         if var it = streamIterator {
           let chunk = try await it.next()
           self.streamIterator = it
-          if chunk == nil {
+          if let chunk {
+            bytesReceived += UInt64(chunk.readableBytes)
+            return chunk
+          } else {
             isFinished = true
+            return nil
           }
-          return chunk
         } else if var it = bodyIterator {
           let chunk = try await it.next()
           self.bodyIterator = it
-          if chunk == nil {
+          if let chunk {
+            bytesReceived += UInt64(chunk.readableBytes)
+            return chunk
+          } else {
             isFinished = true
+            return nil
           }
-          return chunk
         } else {
           isFinished = true
           return nil
         }
+      } catch {
+        guard options.autoResume else {
+          isFinished = true
+          throw error
+        }
+
+        try await resumeDownload(underlyingError: error)
       }
     }
 
-    package var storage: Storage
+    return nil
+  }
 
-    package init(storage: Storage) {
-      self.storage = storage
+  private func resumeDownload(underlyingError: Error) async throws {
+    let currentMetadata = self.metadata ?? ReadObjectMetadata()
+    guard
+      let resumeRange = calculateResumeRange(
+        originalRange: options.range,
+        bytesReceived: bytesReceived,
+        totalSize: currentMetadata.size > 0 ? currentMetadata.size : nil
+      )
+    else {
+      isFinished = true
+      return
     }
 
-    /// Advances to the next `ByteBuffer` chunk in the downloaded object payload stream.
-    public mutating func next() async throws -> NIOCore.ByteBuffer? {
-      try await storage.next()
+    var resumeOptions = options
+    resumeOptions.range = resumeRange
+    if resumeOptions.generation == nil && currentMetadata.generation > 0 {
+      resumeOptions.generation = currentMetadata.generation
+    }
+
+    let httpClient = self.httpClient
+    let bucket = self.bucket
+    let object = self.object
+
+    do {
+      let response = try await retryLoop.run { _ in
+        let request = try await httpClient.buildReadObjectRequest(
+          bucket: bucket, object: object, options: resumeOptions)
+        let resp: _HTTPClientResponse
+        do {
+          resp = try await request.execute()
+        } catch {
+          throw RequestError.io(error)
+        }
+        let statusCode = Int(resp.status.code)
+        if (200..<300).contains(statusCode) {
+          return resp
+        }
+        if resp.isError() {
+          throw await resp.decodeError()
+        }
+        let data = try await resp.data()
+        let message = String(data: data, encoding: .utf8) ?? ""
+        throw DownloadError.unexpectedServerResponse(
+          statusCode: statusCode, message: message)
+      }
+      self.bodyIterator = response.body.makeAsyncIterator()
+      self.streamIterator = nil
+    } catch {
+      isFinished = true
+      if let downloadError = error as? DownloadError {
+        throw downloadError
+      }
+      if let reqError = error as? RequestError {
+        if case .http(let details) = reqError {
+          let message = String(data: details.payload, encoding: .utf8) ?? ""
+          throw DownloadError.unexpectedServerResponse(
+            statusCode: details.http_status_code, message: message)
+        }
+      }
+      throw DownloadError.resumeFailed(
+        bytesReceived: bytesReceived, message: error.localizedDescription)
     }
   }
 
-  /// Creates an asynchronous iterator for iterating over object payload chunks.
-  public func makeAsyncIterator() -> AsyncIterator {
-    let storage = AsyncIterator.Storage(
-      options: options,
-      initialBody: initialBody,
-      stream: stream
-    )
-    return AsyncIterator(storage: storage)
+  package func cancel() {
+    isCancelled = true
+    isFinished = true
+    initialFetchTask?.cancel()
+  }
+
+  fileprivate static func fetchInitial(
+    httpClient: GoogleCloudGax._HTTPClient,
+    bucket: String,
+    object: String,
+    options: ReadObjectOptions,
+    retryLoop: _RetryLoop
+  ) async throws -> (_HTTPClientResponse, ReadObjectMetadata) {
+    if case .bounded(let start, let end) = options.range {
+      guard start <= end else {
+        throw DownloadError.invalidRangeHeader("Range start (\(start)) must be <= end (\(end)).")
+      }
+    }
+    do {
+      return try await retryLoop.run { _ in
+        let request = try await httpClient.buildReadObjectRequest(
+          bucket: bucket, object: object, options: options)
+        let response: _HTTPClientResponse
+        do {
+          response = try await request.execute()
+        } catch {
+          throw RequestError.io(error)
+        }
+        let statusCode = Int(response.status.code)
+        if (200..<300).contains(statusCode) {
+          let metadata = try StorageClient.parseReadObjectMetadata(
+            from: response.headers, bucket: bucket, object: object)
+          return (response, metadata)
+        }
+        if response.isError() {
+          throw await response.decodeError()
+        }
+        let data = try await response.data()
+        let message = String(data: data, encoding: .utf8) ?? ""
+        throw DownloadError.unexpectedServerResponse(
+          statusCode: statusCode, message: message)
+      }
+    } catch let error as RequestError {
+      if case .http(let details) = error {
+        let message = String(data: details.payload, encoding: .utf8) ?? ""
+        throw DownloadError.unexpectedServerResponse(
+          statusCode: details.http_status_code, message: message)
+      } else if case .service(let details) = error {
+        throw DownloadError.unexpectedServerResponse(
+          statusCode: 500, message: details.message)
+      } else {
+        throw error
+      }
+    }
   }
 }
 
 /// Container object returned by `readObject` containing metadata and the streaming body sequence.
-public struct ReadObjectResult: Sendable {
+public struct ReadObjectTask: Sendable {
+  private let coordinator: ReadObjectCoordinator
+
+  package init(coordinator: ReadObjectCoordinator) {
+    self.coordinator = coordinator
+  }
+
   /// Object metadata extracted from initial HTTP response headers.
-  public var metadata: ReadObjectMetadata = .init()
+  public var metadata: ReadObjectMetadata {
+    get async throws {
+      try await coordinator.getMetadata()
+    }
+  }
 
   /// Asynchronous sequence yielding chunks of binary data payload.
-  public var body: ReadObjectSequence = .init()
-
-  /// Creates a new `ReadObjectResult` instance.
-  public init() {}
+  public var body: ReadObjectSequence {
+    ReadObjectSequence(coordinator: coordinator)
+  }
 
-  /// Builder pattern helper to modify configuration in place.
-  public func with(_ config: (inout Self) -> Void) -> Self {
-    var copy = self
-    config(&copy)
-    return copy
+  /// Cancels the ongoing download.
+  public func cancel() {
+    coordinator.cancel()
   }
 }
diff --git a/packages/storage/Sources/GoogleCloudStorage/StorageClient+Download.swift b/packages/storage/Sources/GoogleCloudStorage/StorageClient+Download.swift
index 8b066bf65..c4a70c7d5 100644
--- a/packages/storage/Sources/GoogleCloudStorage/StorageClient+Download.swift
+++ b/packages/storage/Sources/GoogleCloudStorage/StorageClient+Download.swift
@@ -23,48 +23,36 @@ extension StorageClient {
   ///   - bucket: The GCS bucket name.
   ///   - object: The GCS object name.
   ///   - options: Configuration options for the read operation.
-  /// - Returns: A `ReadObjectResult` containing initial object metadata and streaming body sequence.
+  /// - Returns: A `ReadObjectTask` containing initial object metadata and streaming body sequence.
   public func readObject(
     from bucket: String,
     object: String,
     options: ReadObjectOptions = .init()
-  ) async throws -> ReadObjectResult {
-    // TODO(#219): validate range upon construction
-    if case .bounded(let start, let end) = options.range {
-      guard start <= end else {
-        throw DownloadError.invalidRangeHeader("Range start (\(start)) must be <= end (\(end)).")
-      }
-    }
-
-    let request = try await inner.buildReadObjectRequest(
-      bucket: bucket, object: object, options: options)
-    let response = try await request.execute()
-    let statusCode = Int(response.status.code)
-
-    guard (200..<300).contains(statusCode) else {
-      let data = try await response.data()
-      let message = String(data: data, encoding: .utf8) ?? ""
-      throw DownloadError.unexpectedServerResponse(
-        statusCode: statusCode, message: message)
-    }
-
-    let metadata = try Self.parseReadObjectMetadata(
-      from: response.headers, bucket: bucket, object: object)
-
-    let sequence = ReadObjectSequence().with {
-      $0.bucket = bucket
-      $0.object = object
-      $0.options = options
-      $0.metadata = metadata
-      $0.initialBody = response.body
-    }
-    return ReadObjectResult().with {
-      $0.metadata = metadata
-      $0.body = sequence
-    }
+  ) -> ReadObjectTask {
+    let clientOptions = self.options.client
+    let effectiveRetryPolicy =
+      options.retryPolicy ?? self.options.download.retryPolicy
+      ?? clientOptions.retryPolicy
+    let effectiveBackoffPolicy =
+      options.backoffPolicy ?? self.options.download.backoffPolicy ?? clientOptions.backoffPolicy
+    let retryLoop = _RetryLoop(
+      retryPolicy: effectiveRetryPolicy,
+      backoffPolicy: effectiveBackoffPolicy,
+      retryThrottler: clientOptions.retryThrottler,
+      idempotent: true
+    )
+
+    let coordinator = ReadObjectCoordinator(
+      bucket: bucket,
+      object: object,
+      options: options,
+      httpClient: inner,
+      retryLoop: retryLoop
+    )
+    return ReadObjectTask(coordinator: coordinator)
   }
 
-  fileprivate static func parseReadObjectMetadata(
+  package static func parseReadObjectMetadata(
     from headers: NIOHTTP1.HTTPHeaders,
     bucket: String,
     object: String
diff --git a/packages/storage/Sources/GoogleCloudStorage/StorageClientProtocol.swift b/packages/storage/Sources/GoogleCloudStorage/StorageClientProtocol.swift
index 14c8f20ca..2a4f2949f 100644
--- a/packages/storage/Sources/GoogleCloudStorage/StorageClientProtocol.swift
+++ b/packages/storage/Sources/GoogleCloudStorage/StorageClientProtocol.swift
@@ -52,5 +52,5 @@ public protocol StorageClientProtocol {
     from bucket: String,
     object: String,
     options: ReadObjectOptions
-  ) async throws -> ReadObjectResult
+  ) -> ReadObjectTask
 }
diff --git a/packages/storage/Tests/DownloadOptionsTests.swift b/packages/storage/Tests/DownloadOptionsTests.swift
index 8a68daf8c..e8bfd90b9 100644
--- a/packages/storage/Tests/DownloadOptionsTests.swift
+++ b/packages/storage/Tests/DownloadOptionsTests.swift
@@ -96,38 +96,58 @@ import Testing
     #expect(metadata.updated == now)
   }
 
-  @Test func readObjectSequenceAndResponse() async throws {
-    let metadata = ReadObjectMetadata().with {
-      $0.bucket = "bkt"
-      $0.object = "obj"
-    }
-    let sequence = ReadObjectSequence().with {
-      $0.bucket = "bkt"
-      $0.object = "obj"
-    }
+  @Test func calculateResumeRangeScenarios() {
+    // Entire
+    #expect(
+      calculateResumeRange(originalRange: .entire, bytesReceived: 0, totalSize: 1000)
+        == .fromOffset(0))
+    #expect(
+      calculateResumeRange(originalRange: .entire, bytesReceived: 500, totalSize: 1000)
+        == .fromOffset(500))
 
-    #expect(sequence.bucket == "bkt")
-    #expect(sequence.object == "obj")
+    // From offset
+    #expect(
+      calculateResumeRange(originalRange: .fromOffset(100), bytesReceived: 50, totalSize: 1000)
+        == .fromOffset(150))
 
-    var iterator = sequence.makeAsyncIterator()
-    let firstChunk = try await iterator.next()
-    #expect(firstChunk == nil)
+    // Prefix
+    #expect(
+      calculateResumeRange(originalRange: .prefix(100), bytesReceived: 40, totalSize: 1000)
+        == .bounded(start: 40, end: 99))
+    #expect(
+      calculateResumeRange(originalRange: .prefix(100), bytesReceived: 100, totalSize: 1000) == nil)
+    #expect(
+      calculateResumeRange(originalRange: .prefix(100), bytesReceived: 120, totalSize: 1000) == nil)
 
-    let response = ReadObjectResult().with {
-      $0.metadata = metadata
-      $0.body = sequence
-    }
-    #expect(response.metadata.bucket == "bkt")
-    #expect(response.metadata.object == "obj")
-    #expect(response.body.bucket == "bkt")
+    // Bounded
+    #expect(
+      calculateResumeRange(
+        originalRange: .bounded(start: 10, end: 50), bytesReceived: 20, totalSize: 1000)
+        == .bounded(start: 30, end: 50))
+    #expect(
+      calculateResumeRange(
+        originalRange: .bounded(start: 10, end: 50), bytesReceived: 41, totalSize: 1000) == nil)
+
+    // Suffix
+    #expect(
+      calculateResumeRange(originalRange: .suffix(50), bytesReceived: 10, totalSize: 200)
+        == .bounded(start: 160, end: 199))
+    #expect(
+      calculateResumeRange(originalRange: .suffix(50), bytesReceived: 50, totalSize: 200) == nil)
+    #expect(
+      calculateResumeRange(originalRange: .suffix(50), bytesReceived: 10, totalSize: nil)
+        == .fromOffset(10))
   }
 
   @Test func downloadErrorEquality() {
     let err1 = DownloadError.checksumMismatch(expected: "a", actual: "b", algorithm: "crc32c")
     let err2 = DownloadError.checksumMismatch(expected: "a", actual: "b", algorithm: "crc32c")
     let err3 = DownloadError.invalidRangeHeader("bytes=1-0")
+    let err4 = DownloadError.resumeFailed(bytesReceived: 100, message: "failed")
+    let err5 = DownloadError.resumeFailed(bytesReceived: 100, message: "failed")
 
     #expect(err1 == err2)
     #expect(err1 != err3)
+    #expect(err4 == err5)
   }
 }
diff --git a/packages/storage/Tests/DownloadTests.swift b/packages/storage/Tests/DownloadTests.swift
index 3a4064e09..56f3e3eeb 100644
--- a/packages/storage/Tests/DownloadTests.swift
+++ b/packages/storage/Tests/DownloadTests.swift
@@ -27,11 +27,21 @@ import Testing
 
   private static let sampleCsek = sampleKey()
 
-  private func makeClient(registry: MockRegistry) throws -> StorageClient {
+  private func makeClient(
+    registry: MockRegistry,
+    retryPolicy: (any RetryPolicy)? = nil,
+    downloadOptions: ReadObjectOptions? = nil
+  ) throws -> StorageClient {
     let options = StorageClientOptions().with {
       $0.client = .init().with {
         $0.endpoint = registry.endpoint
         $0.credentials = try! Credentials(configuration: .anonymous)
+        if let retryPolicy {
+          $0.retryPolicy = retryPolicy
+        }
+      }
+      if let downloadOptions {
+        $0.download = downloadOptions
       }
     }
     return try StorageClient(options, mock: registry)
@@ -62,19 +72,20 @@ import Testing
     )
 
     let client = try makeClient(registry: registry)
-    let result = try await client.readObject(from: bucket, object: objectName)
-
-    #expect(result.metadata.bucket == bucket)
-    #expect(result.metadata.object == objectName)
-    #expect(result.metadata.size == UInt64(payload.count))
-    #expect(result.metadata.generation == 17123456789)
-    #expect(result.metadata.metageneration == 3)
-    #expect(result.metadata.etag == "\"CPv1234\"")
-    #expect(result.metadata.crc32c == "AdiAvw==")
-    #expect(result.metadata.md5Hash == "N1YvABC==")
-    #expect(result.metadata.contentType == "text/plain; charset=utf-8")
-    #expect(result.metadata.storageClass == "STANDARD")
-    #expect(result.metadata.updated != nil)
+    let result = client.readObject(from: bucket, object: objectName)
+    let metadata = try await result.metadata
+
+    #expect(metadata.bucket == bucket)
+    #expect(metadata.object == objectName)
+    #expect(metadata.size == UInt64(payload.count))
+    #expect(metadata.generation == 17123456789)
+    #expect(metadata.metageneration == 3)
+    #expect(metadata.etag == "\"CPv1234\"")
+    #expect(metadata.crc32c == "AdiAvw==")
+    #expect(metadata.md5Hash == "N1YvABC==")
+    #expect(metadata.contentType == "text/plain; charset=utf-8")
+    #expect(metadata.storageClass == "STANDARD")
+    #expect(metadata.updated != nil)
 
     var downloaded = Data()
     for try await chunk in result.body {
@@ -111,11 +122,12 @@ import Testing
       $0.customerEncryptionKey = csek
     }
 
-    let result = try await client.readObject(from: bucket, object: objectName, options: options)
-    #expect(result.metadata.bucket == bucket)
-    #expect(result.metadata.object == objectName)
-    #expect(result.metadata.size == UInt64(payload.count))
-    #expect(result.metadata.generation == 42)
+    let result = client.readObject(from: bucket, object: objectName, options: options)
+    let metadata = try await result.metadata
+    #expect(metadata.bucket == bucket)
+    #expect(metadata.object == objectName)
+    #expect(metadata.size == UInt64(payload.count))
+    #expect(metadata.generation == 42)
 
     let lastReq = registry.lastRequest(for: downloadUrl)
     #expect(lastReq != nil)
@@ -212,8 +224,9 @@ import Testing
     )
 
     let client = try makeClient(registry: registry)
-    let result = try await client.readObject(from: bucket, object: objectName, options: options)
-    #expect(result.metadata.size == 4)
+    let result = client.readObject(from: bucket, object: objectName, options: options)
+    let metadata = try await result.metadata
+    #expect(metadata.size == 4)
 
     let lastReq = registry.lastRequest(for: downloadUrl)
     #expect(lastReq != nil)
@@ -249,7 +262,7 @@ import Testing
     }
 
     let err = await expectError(DownloadError.self) {
-      try await client.readObject(from: bucket, object: objectName, options: options)
+      try await client.readObject(from: bucket, object: objectName, options: options).metadata
     }
 
     if case .unexpectedServerResponse(let statusCode, let message) = err {
@@ -286,10 +299,11 @@ import Testing
     )
 
     let client = try makeClient(registry: registry)
-    let result = try await client.readObject(from: bucket, object: objectName)
+    let result = client.readObject(from: bucket, object: objectName)
+    let metadata = try await result.metadata
 
-    #expect(result.metadata.bucket == bucket)
-    #expect(result.metadata.object == objectName)
+    #expect(metadata.bucket == bucket)
+    #expect(metadata.object == objectName)
 
     var downloaded = Data()
     for try await chunk in result.body {
@@ -313,7 +327,7 @@ import Testing
     let client = try makeClient(registry: registry)
 
     let err = await expectError(DownloadError.self) {
-      try await client.readObject(from: bucket, object: objectName)
+      try await client.readObject(from: bucket, object: objectName).metadata
     }
 
     if case .unexpectedServerResponse(let statusCode, let message) = err {
@@ -351,9 +365,10 @@ import Testing
       $0.range = .fromOffset(10)
     }
 
-    let result = try await client.readObject(from: bucket, object: objectName, options: options)
-    #expect(result.metadata.size == 50)
-    #expect(result.metadata.generation == 123)
+    let result = client.readObject(from: bucket, object: objectName, options: options)
+    let metadata = try await result.metadata
+    #expect(metadata.size == 50)
+    #expect(metadata.generation == 123)
 
     let lastReq = registry.lastRequest(for: downloadUrl)
     #expect(lastReq?.value(forHTTPHeaderField: "Range") == "bytes=10-")
@@ -392,8 +407,9 @@ import Testing
       $0.range = .prefix(20)
     }
 
-    let result = try await client.readObject(from: bucket, object: objectName, options: options)
-    #expect(result.metadata.size == 50)
+    let result = client.readObject(from: bucket, object: objectName, options: options)
+    let metadata = try await result.metadata
+    #expect(metadata.size == 50)
 
     let lastReq = registry.lastRequest(for: downloadUrl)
     #expect(lastReq?.value(forHTTPHeaderField: "Range") == "bytes=0-19")
@@ -432,8 +448,9 @@ import Testing
       $0.range = .suffix(15)
     }
 
-    let result = try await client.readObject(from: bucket, object: objectName, options: options)
-    #expect(result.metadata.size == 50)
+    let result = client.readObject(from: bucket, object: objectName, options: options)
+    let metadata = try await result.metadata
+    #expect(metadata.size == 50)
 
     let lastReq = registry.lastRequest(for: downloadUrl)
     #expect(lastReq?.value(forHTTPHeaderField: "Range") == "bytes=-15")
@@ -472,8 +489,9 @@ import Testing
       $0.range = .bounded(start: 10, end: 29)
     }
 
-    let result = try await client.readObject(from: bucket, object: objectName, options: options)
-    #expect(result.metadata.size == 50)
+    let result = client.readObject(from: bucket, object: objectName, options: options)
+    let metadata = try await result.metadata
+    #expect(metadata.size == 50)
 
     let lastReq = registry.lastRequest(for: downloadUrl)
     #expect(lastReq?.value(forHTTPHeaderField: "Range") == "bytes=10-29")
@@ -505,7 +523,8 @@ import Testing
 
     _ = try await client.readObject(
       from: bucket, object: objectName,
-      options: ReadObjectOptions().with { $0.range = ReadObjectRange(10...29) })
+      options: ReadObjectOptions().with { $0.range = ReadObjectRange(10...29) }
+    ).metadata
     #expect(
       registry.lastRequest(for: downloadUrl)?.value(forHTTPHeaderField: "Range") == "bytes=10-29")
   }
@@ -519,6 +538,7 @@ import Testing
 
     do {
       _ = try await client.readObject(from: "test-bucket", object: "test.txt", options: options)
+        .metadata
       Issue.record("Expected invalidRangeHeader error to be thrown")
     } catch let DownloadError.invalidRangeHeader(msg) {
       #expect(msg.contains("30"))
@@ -548,13 +568,14 @@ import Testing
       for: prefixUrl
     )
 
-    let prefixResult = try await client.readObject(
+    let prefixResult = client.readObject(
       from: bucket,
       object: prefixObject,
       options: ReadObjectOptions().with { $0.range = .prefix(0) }
     )
-    #expect(prefixResult.metadata.size == 50)
-    #expect(prefixResult.metadata.generation == 123)
+    let prefixMeta = try await prefixResult.metadata
+    #expect(prefixMeta.size == 50)
+    #expect(prefixMeta.generation == 123)
     #expect(
       registry.lastRequest(for: prefixUrl)?.value(forHTTPHeaderField: "Range") == "bytes=0-0")
 
@@ -579,13 +600,14 @@ import Testing
       for: suffixUrl
     )
 
-    let suffixResult = try await client.readObject(
+    let suffixResult = client.readObject(
       from: bucket,
       object: suffixObject,
       options: ReadObjectOptions().with { $0.range = .suffix(0) }
     )
-    #expect(suffixResult.metadata.size == 50)
-    #expect(suffixResult.metadata.generation == 123)
+    let suffixMeta = try await suffixResult.metadata
+    #expect(suffixMeta.size == 50)
+    #expect(suffixMeta.generation == 123)
     #expect(
       registry.lastRequest(for: suffixUrl)?.value(forHTTPHeaderField: "Range") == "bytes=-0")
 
@@ -606,7 +628,7 @@ import Testing
         from: bucket,
         object: "nonexistent.txt",
         options: ReadObjectOptions().with { $0.range = .prefix(0) }
-      )
+      ).metadata
       Issue.record("Expected unexpectedServerResponse error to be thrown for 404")
     } catch DownloadError.unexpectedServerResponse(let statusCode, _) {
       #expect(statusCode == 404)
@@ -653,12 +675,13 @@ import Testing
       $0.enableDecompressiveTranscoding = false
     }
 
-    let result = try await client.readObject(from: bucket, object: objectName, options: options)
+    let result = client.readObject(from: bucket, object: objectName, options: options)
+    let metadata = try await result.metadata
 
-    #expect(result.metadata.bucket == bucket)
-    #expect(result.metadata.object == objectName)
-    #expect(result.metadata.contentEncoding == "gzip")
-    #expect(result.metadata.size == UInt64(payload.count))
+    #expect(metadata.bucket == bucket)
+    #expect(metadata.object == objectName)
+    #expect(metadata.contentEncoding == "gzip")
+    #expect(metadata.size == UInt64(payload.count))
 
     let lastReq = registry.lastRequest(for: downloadUrl)
     #expect(lastReq != nil)
@@ -698,11 +721,12 @@ import Testing
       $0.enableDecompressiveTranscoding = true
     }
 
-    let result = try await client.readObject(from: bucket, object: objectName, options: options)
+    let result = client.readObject(from: bucket, object: objectName, options: options)
+    let metadata = try await result.metadata
 
-    #expect(result.metadata.bucket == bucket)
-    #expect(result.metadata.object == objectName)
-    #expect(result.metadata.size == UInt64(payload.count))
+    #expect(metadata.bucket == bucket)
+    #expect(metadata.object == objectName)
+    #expect(metadata.size == UInt64(payload.count))
 
     let lastReq = registry.lastRequest(for: downloadUrl)
     #expect(lastReq != nil)
@@ -740,10 +764,11 @@ import Testing
     )
 
     let client = try makeClient(registry: registry)
-    let result = try await client.readObject(from: bucket, object: objectName)
+    let result = client.readObject(from: bucket, object: objectName)
+    let metadata = try await result.metadata
 
-    #expect(result.metadata.size == UInt64(fullPayload.count))
-    #expect(result.metadata.generation == 999)
+    #expect(metadata.size == UInt64(fullPayload.count))
+    #expect(metadata.generation == 999)
 
     var receivedChunks: [Data] = []
     for try await chunk in result.body {
@@ -756,4 +781,518 @@ import Testing
     #expect(receivedChunks[2] == chunk3)
     #expect(receivedChunks.reduce(Data(), +) == fullPayload)
   }
+
+  @Test func downloadObjectTransientFailureRetriesAndSucceeds() async throws {
+    let registry = MockRegistry.create()
+    let bucket = "test-bucket"
+    let objectName = "test-retry.txt"
+    let payload = Data("Download retry success payload".utf8)
+
+    let downloadUrl = registry.url("/storage/v1/b/\(bucket)/o/\(objectName)?alt=media")
+
+    registry.register(
+      response: .success(statusCode: 503, data: Data("Service Unavailable".utf8), headers: nil),
+      for: downloadUrl
+    )
+    registry.register(
+      response: .success(
+        statusCode: 200,
+        data: payload,
+        headers: [
+          "Content-Length": String(payload.count),
+          "x-goog-generation": "100",
+        ]
+      ),
+      for: downloadUrl
+    )
+
+    let client = try makeClient(
+      registry: registry, retryPolicy: BaseRetryPolicy().withAttemptLimit(3))
+    let result = client.readObject(from: bucket, object: objectName)
+    let metadata = try await result.metadata
+
+    #expect(metadata.generation == 100)
+    var downloaded = Data()
+    for try await chunk in result.body {
+      downloaded.append(contentsOf: chunk.readableBytesView)
+    }
+    #expect(downloaded == payload)
+
+    let requests = registry.recordedRequests()
+    #expect(requests.count == 2)
+  }
+
+  @Test func downloadObjectTransientFailureWithNeverRetryFails() async throws {
+    let registry = MockRegistry.create()
+    let bucket = "test-bucket"
+    let objectName = "test-never-retry.txt"
+
+    let downloadUrl = registry.url("/storage/v1/b/\(bucket)/o/\(objectName)?alt=media")
+
+    registry.register(
+      response: .success(statusCode: 503, data: Data("Service Unavailable".utf8), headers: nil),
+      for: downloadUrl
+    )
+
+    let client = try makeClient(registry: registry, retryPolicy: NeverRetry())
+
+    let err = await expectError(DownloadError.self) {
+      try await client.readObject(from: bucket, object: objectName).metadata
+    }
+    #expect(err != nil)
+
+    let requests = registry.recordedRequests()
+    #expect(requests.count == 1)
+  }
+
+  @Test func downloadObjectWithCustomReadObjectOptionsRetryPolicyOverridesClient() async throws {
+    let registry = MockRegistry.create()
+    let bucket = "test-bucket"
+    let objectName = "test-override-retry.txt"
+
+    let downloadUrl = registry.url("/storage/v1/b/\(bucket)/o/\(objectName)?alt=media")
+
+    registry.register(
+      response: .success(statusCode: 503, data: Data("Service Unavailable".utf8), headers: nil),
+      for: downloadUrl
+    )
+
+    let client = try makeClient(registry: registry)
+    let options = ReadObjectOptions().with {
+      $0.retryPolicy = NeverRetry()
+    }
+
+    let err = await expectError(DownloadError.self) {
+      try await client.readObject(from: bucket, object: objectName, options: options).metadata
+    }
+    #expect(err != nil)
+
+    let requests = registry.recordedRequests()
+    #expect(requests.count == 1)
+  }
+
+  @Test func downloadObjectWithClientDownloadOptionsRetryPolicyOverridesDefault() async throws {
+    let registry = MockRegistry.create()
+    let bucket = "test-bucket"
+    let objectName = "test-client-override.txt"
+
+    let downloadUrl = registry.url("/storage/v1/b/\(bucket)/o/\(objectName)?alt=media")
+
+    registry.register(
+      response: .success(statusCode: 503, data: Data("Service Unavailable".utf8), headers: nil),
+      for: downloadUrl
+    )
+
+    let client = try makeClient(
+      registry: registry,
+      downloadOptions: ReadObjectOptions().with { $0.retryPolicy = NeverRetry() }
+    )
+
+    let err = await expectError(DownloadError.self) {
+      try await client.readObject(from: bucket, object: objectName).metadata
+    }
+    #expect(err != nil)
+
+    let requests = registry.recordedRequests()
+    #expect(requests.count == 1)
+  }
+
+  @Test func downloadObjectStreamingTransientFailureResumesFromLatestByte() async throws {
+    let registry = MockRegistry.create()
+    let bucket = "test-bucket"
+    let objectName = "streaming-resume.bin"
+    let chunk1 = Data("FirstPart-".utf8)
+    let chunk2 = Data("SecondPart".utf8)
+    let fullPayload = chunk1 + chunk2
+
+    let initialUrl = registry.url("/storage/v1/b/\(bucket)/o/\(objectName)?alt=media")
+    let resumeUrl = registry.url(
+      "/storage/v1/b/\(bucket)/o/\(objectName)?alt=media&generation=888")
+
+    registry.register(
+      response: .stream(
+        statusCode: 200,
+        chunks: [chunk1],
+        error: MockNetworkError(),
+        headers: [
+          "Content-Length": String(fullPayload.count),
+          "x-goog-generation": "888",
+        ]
+      ),
+      for: initialUrl
+    )
+
+    registry.register(
+      response: .stream(
+        statusCode: 206,
+        chunks: [chunk2],
+        headers: [
+          "Content-Range": "bytes \(chunk1.count)-\(fullPayload.count - 1)/\(fullPayload.count)",
+          "Content-Length": String(chunk2.count),
+          "x-goog-generation": "888",
+        ]
+      ),
+      for: resumeUrl
+    )
+
+    let client = try makeClient(registry: registry)
+    let result = client.readObject(from: bucket, object: objectName)
+    let metadata = try await result.metadata
+
+    #expect(metadata.size == UInt64(fullPayload.count))
+    #expect(metadata.generation == 888)
+
+    var receivedData = Data()
+    for try await chunk in result.body {
+      receivedData.append(contentsOf: chunk.readableBytesView)
+    }
+
+    #expect(receivedData == fullPayload)
+
+    let requests = registry.recordedRequests()
+    #expect(requests.count == 2)
+    #expect(requests[0].value(forHTTPHeaderField: "Range") == nil)
+    #expect(requests[1].value(forHTTPHeaderField: "Range") == "bytes=\(chunk1.count)-")
+  }
+
+  @Test func downloadObjectStreamingMultipleTransientFailuresResumesAndRecovers() async throws {
+    let registry = MockRegistry.create()
+    let bucket = "test-bucket"
+    let objectName = "multi-resume.bin"
+    let chunk1 = Data("1111111111".utf8)
+    let chunk2 = Data("2222222222".utf8)
+    let chunk3 = Data("3333333333".utf8)
+    let fullPayload = chunk1 + chunk2 + chunk3
+
+    let initialUrl = registry.url("/storage/v1/b/\(bucket)/o/\(objectName)?alt=media")
+    let resumeUrl = registry.url(
+      "/storage/v1/b/\(bucket)/o/\(objectName)?alt=media&generation=777")
+
+    // Attempt 1: Yields chunk1, then network fails
+    registry.register(
+      response: .stream(
+        statusCode: 200,
+        chunks: [chunk1],
+        error: MockNetworkError(),
+        headers: [
+          "Content-Length": String(fullPayload.count),
+          "x-goog-generation": "777",
+        ]
+      ),
+      for: initialUrl
+    )
+
+    // Attempt 2 (Resume at 10): Yields chunk2, then network fails again
+    registry.register(
+      response: .stream(
+        statusCode: 206,
+        chunks: [chunk2],
+        error: MockNetworkError(),
+        headers: [
+          "Content-Range": "bytes 10-\(fullPayload.count - 1)/\(fullPayload.count)",
+          "Content-Length": String(chunk2.count + chunk3.count),
+          "x-goog-generation": "777",
+        ]
+      ),
+      for: resumeUrl
+    )
+
+    // Attempt 3 (Resume at 20): Yields chunk3 and finishes cleanly
+    registry.register(
+      response: .stream(
+        statusCode: 206,
+        chunks: [chunk3],
+        headers: [
+          "Content-Range": "bytes 20-\(fullPayload.count - 1)/\(fullPayload.count)",
+          "Content-Length": String(chunk3.count),
+          "x-goog-generation": "777",
+        ]
+      ),
+      for: resumeUrl
+    )
+
+    let client = try makeClient(registry: registry)
+    let result = client.readObject(from: bucket, object: objectName)
+
+    var receivedData = Data()
+    for try await chunk in result.body {
+      receivedData.append(contentsOf: chunk.readableBytesView)
+    }
+
+    #expect(receivedData == fullPayload)
+
+    let requests = registry.recordedRequests()
+    #expect(requests.count == 3)
+    #expect(requests[0].value(forHTTPHeaderField: "Range") == nil)
+    #expect(requests[1].value(forHTTPHeaderField: "Range") == "bytes=10-")
+    #expect(requests[2].value(forHTTPHeaderField: "Range") == "bytes=20-")
+  }
+
+  @Test func downloadObjectStreamingWithBoundedRangeResumesFromOffset() async throws {
+    let registry = MockRegistry.create()
+    let bucket = "test-bucket"
+    let objectName = "bounded-resume.bin"
+    let chunk1 = Data("0123456789".utf8)
+    let chunk2 = Data("abcdefghij".utf8)
+    let expectedPayload = chunk1 + chunk2
+
+    let initialUrl = registry.url("/storage/v1/b/\(bucket)/o/\(objectName)?alt=media")
+    let resumeUrl = registry.url(
+      "/storage/v1/b/\(bucket)/o/\(objectName)?alt=media&generation=555")
+
+    registry.register(
+      response: .stream(
+        statusCode: 206,
+        chunks: [chunk1],
+        error: MockNetworkError(),
+        headers: [
+          "Content-Range": "bytes 10-29/100",
+          "Content-Length": "20",
+          "x-goog-generation": "555",
+        ]
+      ),
+      for: initialUrl
+    )
+
+    registry.register(
+      response: .stream(
+        statusCode: 206,
+        chunks: [chunk2],
+        headers: [
+          "Content-Range": "bytes 20-29/100",
+          "Content-Length": "10",
+          "x-goog-generation": "555",
+        ]
+      ),
+      for: resumeUrl
+    )
+
+    let client = try makeClient(registry: registry)
+    let options = ReadObjectOptions().with {
+      $0.range = .bounded(start: 10, end: 29)
+    }
+
+    let result = client.readObject(from: bucket, object: objectName, options: options)
+
+    var receivedData = Data()
+    for try await chunk in result.body {
+      receivedData.append(contentsOf: chunk.readableBytesView)
+    }
+
+    #expect(receivedData == expectedPayload)
+
+    let requests = registry.recordedRequests()
+    #expect(requests.count == 2)
+    #expect(requests[0].value(forHTTPHeaderField: "Range") == "bytes=10-29")
+    #expect(requests[1].value(forHTTPHeaderField: "Range") == "bytes=20-29")
+  }
+
+  @Test func downloadObjectStreamingWithAutoResumeDisabledThrowsOnStreamInterruption() async throws
+  {
+    let registry = MockRegistry.create()
+    let bucket = "test-bucket"
+    let objectName = "no-resume.bin"
+    let chunk1 = Data("FirstPart-".utf8)
+
+    let initialUrl = registry.url("/storage/v1/b/\(bucket)/o/\(objectName)?alt=media")
+
+    registry.register(
+      response: .stream(
+        statusCode: 200,
+        chunks: [chunk1],
+        error: MockNetworkError(),
+        headers: [
+          "Content-Length": "100",
+          "x-goog-generation": "333",
+        ]
+      ),
+      for: initialUrl
+    )
+
+    let client = try makeClient(registry: registry)
+    let options = ReadObjectOptions().with {
+      $0.autoResume = false
+    }
+
+    let result = client.readObject(from: bucket, object: objectName, options: options)
+
+    do {
+      for try await _ in result.body {}
+      Issue.record("Expected error to be thrown when autoResume is false")
+    } catch is MockNetworkError {
+      // Expected
+    } catch {
+      Issue.record("Expected MockNetworkError, but got \(error)")
+    }
+
+    let requests = registry.recordedRequests()
+    #expect(requests.count == 1)
+  }
+
+  @Test func downloadObjectStreamingResumePermanentErrorThrows() async throws {
+    let registry = MockRegistry.create()
+    let bucket = "test-bucket"
+    let objectName = "resume-404.bin"
+    let chunk1 = Data("InitialData".utf8)
+
+    let initialUrl = registry.url("/storage/v1/b/\(bucket)/o/\(objectName)?alt=media")
+    let resumeUrl = registry.url(
+      "/storage/v1/b/\(bucket)/o/\(objectName)?alt=media&generation=222")
+
+    registry.register(
+      response: .stream(
+        statusCode: 200,
+        chunks: [chunk1],
+        error: MockNetworkError(),
+        headers: [
+          "Content-Length": "50",
+          "x-goog-generation": "222",
+        ]
+      ),
+      for: initialUrl
+    )
+
+    registry.register(
+      response: .success(statusCode: 404, data: Data("Object deleted".utf8), headers: nil),
+      for: resumeUrl
+    )
+
+    let client = try makeClient(registry: registry)
+    let result = client.readObject(from: bucket, object: objectName)
+
+    do {
+      for try await _ in result.body {}
+      Issue.record("Expected error when resume fails with 404")
+    } catch DownloadError.unexpectedServerResponse(let statusCode, let message) {
+      #expect(statusCode == 404)
+      #expect(message == "Object deleted")
+    } catch {
+      Issue.record("Expected unexpectedServerResponse 404, got \(error)")
+    }
+  }
+
+  @Test func downloadObjectStreamingWithCustomerSuppliedEncryptionKeyPreservesHeadersOnResume()
+    async throws
+  {
+    let registry = MockRegistry.create()
+    let bucket = "csek-bucket"
+    let objectName = "csek-resume.bin"
+    let chunk1 = Data("encrypted-part1".utf8)
+    let chunk2 = Data("encrypted-part2".utf8)
+    let fullPayload = chunk1 + chunk2
+    let csek = Self.sampleCsek
+
+    let initialUrl = registry.url("/storage/v1/b/\(bucket)/o/\(objectName)?alt=media")
+    let resumeUrl = registry.url(
+      "/storage/v1/b/\(bucket)/o/\(objectName)?alt=media&generation=111")
+
+    registry.register(
+      response: .stream(
+        statusCode: 200,
+        chunks: [chunk1],
+        error: MockNetworkError(),
+        headers: [
+          "Content-Length": String(fullPayload.count),
+          "x-goog-generation": "111",
+        ]
+      ),
+      for: initialUrl
+    )
+
+    registry.register(
+      response: .stream(
+        statusCode: 206,
+        chunks: [chunk2],
+        headers: [
+          "Content-Range": "bytes \(chunk1.count)-\(fullPayload.count - 1)/\(fullPayload.count)",
+          "Content-Length": String(chunk2.count),
+          "x-goog-generation": "111",
+        ]
+      ),
+      for: resumeUrl
+    )
+
+    let client = try makeClient(registry: registry)
+    let options = ReadObjectOptions().with {
+      $0.customerEncryptionKey = csek
+    }
+
+    let result = client.readObject(from: bucket, object: objectName, options: options)
+
+    var receivedData = Data()
+    for try await chunk in result.body {
+      receivedData.append(contentsOf: chunk.readableBytesView)
+    }
+
+    #expect(receivedData == fullPayload)
+
+    let requests = registry.recordedRequests()
+    #expect(requests.count == 2)
+    #expect(requests[1].value(forHTTPHeaderField: "x-goog-encryption-algorithm") == "AES256")
+    #expect(requests[1].value(forHTTPHeaderField: "x-goog-encryption-key") == csek.keyBase64)
+    #expect(
+      requests[1].value(forHTTPHeaderField: "x-goog-encryption-key-sha256") == csek.keyHashBase64)
+  }
+
+  @Test func downloadObjectDeferredExecutionOnlyAwaitsMetadata() async throws {
+    let registry = MockRegistry.create()
+    let bucket = "test-bucket"
+    let objectName = "meta-only.txt"
+    let payload = Data("Only read metadata".utf8)
+
+    let downloadUrl = registry.url("/storage/v1/b/\(bucket)/o/\(objectName)?alt=media")
+
+    registry.register(
+      response: .success(
+        statusCode: 200,
+        data: payload,
+        headers: [
+          "Content-Length": String(payload.count),
+          "x-goog-generation": "999",
+        ]
+      ),
+      for: downloadUrl
+    )
+
+    let client = try makeClient(registry: registry)
+    let result = client.readObject(from: bucket, object: objectName)
+    let metadata = try await result.metadata
+
+    #expect(metadata.size == UInt64(payload.count))
+    #expect(metadata.generation == 999)
+
+    // Result body was never consumed, download finishes or cancels cleanly
+    result.cancel()
+  }
+
+  @Test func downloadObjectDeferredExecutionOnlyConsumesBody() async throws {
+    let registry = MockRegistry.create()
+    let bucket = "test-bucket"
+    let objectName = "body-only.txt"
+    let payload = Data("Direct stream consumption without metadata read".utf8)
+
+    let downloadUrl = registry.url("/storage/v1/b/\(bucket)/o/\(objectName)?alt=media")
+
+    registry.register(
+      response: .success(
+        statusCode: 200,
+        data: payload,
+        headers: [
+          "Content-Length": String(payload.count),
+          "x-goog-generation": "888",
+        ]
+      ),
+      for: downloadUrl
+    )
+
+    let client = try makeClient(registry: registry)
+    let result = client.readObject(from: bucket, object: objectName)
+
+    var receivedData = Data()
+    for try await chunk in result.body {
+      receivedData.append(contentsOf: chunk.readableBytesView)
+    }
+
+    #expect(receivedData == payload)
+  }
 }
diff --git a/packages/storage/Tests/IntegrationTests/StorageClientIntegrationTests.swift b/packages/storage/Tests/IntegrationTests/StorageClientIntegrationTests.swift
index 333be6bd8..42b752f9f 100644
--- a/packages/storage/Tests/IntegrationTests/StorageClientIntegrationTests.swift
+++ b/packages/storage/Tests/IntegrationTests/StorageClientIntegrationTests.swift
@@ -70,11 +70,12 @@ import Testing
       #expect(uploadedObject.bucket == bucketName)
       #expect(uploadedObject.name == objectName)
 
-      let result = try await storage.readObject(from: bucketName, object: objectName)
-      #expect(result.metadata.bucket == bucketName)
-      #expect(result.metadata.object == objectName)
-      #expect(result.metadata.size == UInt64(data.count))
-      #expect(result.metadata.generation == UInt64(uploadedObject.generation))
+      let result = storage.readObject(from: bucketName, object: objectName)
+      let metadata = try await result.metadata
+      #expect(metadata.bucket == bucketName)
+      #expect(metadata.object == objectName)
+      #expect(metadata.size == UInt64(data.count))
+      #expect(metadata.generation == UInt64(uploadedObject.generation))
 
       var downloadedData = Data()
       for try await chunk in result.body {
@@ -84,7 +85,7 @@ import Testing
       let downloadedString = String(data: downloadedData, encoding: .utf8)
       #expect(downloadedString == content)
 
-      print("File download integration test successful: \(result.metadata)")
+      print("File download integration test successful: \(metadata)")
     }
 
     @Test func testFailedDownloadPrecondition() async throws {
@@ -108,6 +109,7 @@ import Testing
 
       do {
         _ = try await storage.readObject(from: bucketName, object: objectName, options: options)
+          .metadata
         Issue.record("Expected download to fail with 412 Precondition Failed, but it succeeded")
       } catch DownloadError.unexpectedServerResponse(let statusCode, let message) {
         #expect(statusCode == 412)
@@ -139,11 +141,12 @@ import Testing
       #expect(uploadedObject.name == objectName)
       #expect(uploadedObject.size == Int64(data.count))
 
-      let result = try await storage.readObject(from: bucketName, object: objectName)
-      #expect(result.metadata.bucket == bucketName)
-      #expect(result.metadata.object == objectName)
-      #expect(result.metadata.size == UInt64(data.count))
-      #expect(result.metadata.generation == UInt64(uploadedObject.generation))
+      let result = storage.readObject(from: bucketName, object: objectName)
+      let metadata = try await result.metadata
+      #expect(metadata.bucket == bucketName)
+      #expect(metadata.object == objectName)
+      #expect(metadata.size == UInt64(data.count))
+      #expect(metadata.generation == UInt64(uploadedObject.generation))
 
       var downloadedData = Data()
       for try await chunk in result.body {
@@ -520,7 +523,7 @@ import Testing
 
       // 2. Attempt download without CSEK key (GCS rejects with 400 Bad Request)
       do {
-        _ = try await storage.readObject(from: bucketName, object: objectName)
+        _ = try await storage.readObject(from: bucketName, object: objectName).metadata
         Issue.record("Expected download without CSEK key to fail")
       } catch DownloadError.unexpectedServerResponse(let statusCode, _) {
         #expect(statusCode == 400)
@@ -532,12 +535,13 @@ import Testing
       let downloadOptions = ReadObjectOptions().with {
         $0.customerEncryptionKey = csek
       }
-      let result = try await storage.readObject(
+      let result = storage.readObject(
         from: bucketName, object: objectName, options: downloadOptions)
-      #expect(result.metadata.bucket == bucketName)
-      #expect(result.metadata.object == objectName)
-      #expect(result.metadata.size == UInt64(data.count))
-      #expect(result.metadata.generation == UInt64(uploadedObject.generation))
+      let metadata = try await result.metadata
+      #expect(metadata.bucket == bucketName)
+      #expect(metadata.object == objectName)
+      #expect(metadata.size == UInt64(data.count))
+      #expect(metadata.generation == UInt64(uploadedObject.generation))
 
       var downloadedData = Data()
       for try await chunk in result.body {
@@ -547,7 +551,7 @@ import Testing
       let downloadedString = String(data: downloadedData, encoding: .utf8)
       #expect(downloadedString == content)
 
-      print("CSEK upload and download successful: \(result.metadata)")
+      print("CSEK upload and download successful: \(metadata)")
     }
 
     @Test func testCSEKSimpleUpload() async throws {
@@ -608,9 +612,10 @@ import Testing
       let downloadOptions = ReadObjectOptions().with {
         $0.customerEncryptionKey = csek
       }
-      let result = try await storage.readObject(
+      let result = storage.readObject(
         from: bucketName, object: objectName, options: downloadOptions)
-      #expect(result.metadata.size == UInt64(fileSize))
+      let metadata = try await result.metadata
+      #expect(metadata.size == UInt64(fileSize))
 
       var downloadedData = Data()
       for try await chunk in result.body {
@@ -771,10 +776,11 @@ import Testing
       let downloadOptions = ReadObjectOptions().with {
         $0.enableDecompressiveTranscoding = true
       }
-      let result = try await storage.readObject(
+      let result = storage.readObject(
         from: bucketName, object: objectName, options: downloadOptions)
-      #expect(result.metadata.bucket == bucketName)
-      #expect(result.metadata.object == objectName)
+      let metadata = try await result.metadata
+      #expect(metadata.bucket == bucketName)
+      #expect(metadata.object == objectName)
 
       var downloadedData = Data()
       for try await chunk in result.body {
@@ -785,7 +791,7 @@ import Testing
       let downloadedString = String(data: downloadedData, encoding: .utf8)
       #expect(downloadedString == Self.rawGzipContent)
 
-      print("Gzip allow decompressive transcoding integration test successful: \(result.metadata)")
+      print("Gzip allow decompressive transcoding integration test successful: \(metadata)")
     }
 
     @Test func testDownloadGzipPreventTranscodingViaRequestHeader() async throws {
@@ -811,11 +817,12 @@ import Testing
       let downloadOptions = ReadObjectOptions().with {
         $0.enableDecompressiveTranscoding = false
       }
-      let result = try await storage.readObject(
+      let result = storage.readObject(
         from: bucketName, object: objectName, options: downloadOptions)
-      #expect(result.metadata.bucket == bucketName)
-      #expect(result.metadata.object == objectName)
-      #expect(result.metadata.contentEncoding == "gzip")
+      let metadata = try await result.metadata
+      #expect(metadata.bucket == bucketName)
+      #expect(metadata.object == objectName)
+      #expect(metadata.contentEncoding == "gzip")
 
       var downloadedData = Data()
       for try await chunk in result.body {
@@ -823,10 +830,10 @@ import Testing
       }
       // Downloader receives the original gzip-compressed file
       #expect(downloadedData == Self.compressedGzipData)
-      #expect(result.metadata.size == UInt64(Self.compressedGzipData.count))
+      #expect(metadata.size == UInt64(Self.compressedGzipData.count))
 
       print(
-        "Gzip prevent transcoding via request header integration test successful: \(result.metadata)"
+        "Gzip prevent transcoding via request header integration test successful: \(metadata)"
       )
     }
 
@@ -852,10 +859,11 @@ import Testing
       #expect(uploadedObject.cacheControl == "no-transform")
 
       // Standard download request without special options
-      let result = try await storage.readObject(from: bucketName, object: objectName)
-      #expect(result.metadata.bucket == bucketName)
-      #expect(result.metadata.object == objectName)
-      #expect(result.metadata.contentEncoding == "gzip")
+      let result = storage.readObject(from: bucketName, object: objectName)
+      let metadata = try await result.metadata
+      #expect(metadata.bucket == bucketName)
+      #expect(metadata.object == objectName)
+      #expect(metadata.contentEncoding == "gzip")
 
       var downloadedData = Data()
       for try await chunk in result.body {
@@ -863,10 +871,10 @@ import Testing
       }
       // Downloader receives the original gzip-compressed file because of Cache-Control: no-transform
       #expect(downloadedData == Self.compressedGzipData)
-      #expect(result.metadata.size == UInt64(Self.compressedGzipData.count))
+      #expect(metadata.size == UInt64(Self.compressedGzipData.count))
 
       print(
-        "Gzip prevent transcoding via Cache-Control no-transform integration test successful: \(result.metadata)"
+        "Gzip prevent transcoding via Cache-Control no-transform integration test successful: \(metadata)"
       )
     }
   }
@@ -920,11 +928,12 @@ import Testing
       let options = ReadObjectOptions().with {
         $0.range = range
       }
-      let result = try await storage.readObject(
+      let result = storage.readObject(
         from: fixture.bucketName, object: fixture.objectName, options: options)
+      let metadata = try await result.metadata
 
-      #expect(result.metadata.size == fixture.totalSize)
-      #expect(result.metadata.generation == UInt64(fixture.uploadedObject.generation))
+      #expect(metadata.size == fixture.totalSize)
+      #expect(metadata.generation == UInt64(fixture.uploadedObject.generation))
 
       var downloadedData = Data()
       for try await chunk in result.body {
@@ -940,11 +949,12 @@ import Testing
       let options = ReadObjectOptions().with {
         $0.range = .prefix(0)
       }
-      let result = try await storage.readObject(
+      let result = storage.readObject(
         from: fixture.bucketName, object: fixture.objectName, options: options)
+      let metadata = try await result.metadata
 
-      #expect(result.metadata.size == fixture.totalSize)
-      #expect(result.metadata.generation == UInt64(fixture.uploadedObject.generation))
+      #expect(metadata.size == fixture.totalSize)
+      #expect(metadata.generation == UInt64(fixture.uploadedObject.generation))
 
       var downloadedData = Data()
       for try await chunk in result.body {
@@ -972,7 +982,7 @@ import Testing
           from: bucketName,
           object: "non-existent-\(UUID().uuidString).txt",
           options: options
-        )
+        ).metadata
         Issue.record("Expected reading non-existent object to throw 404")
       } catch DownloadError.unexpectedServerResponse(let statusCode, _) {
         #expect(statusCode == 404)