@@ -72,6 +72,7 @@ class ChunkAttemptCallable<ResponseT> implements Callable<ChunkUploadResponse<Re
   private final RewindableStreamBuffer buffer;
   private final String uploadUrl;
   private final ApiCallContext originalCallContext;
+  private final UploadProgressTracker progressTracker;
   private final long deadlineNanos;
   private final ApiClock clock;
 
@@ -90,7 +91,8 @@ class ChunkAttemptCallable<ResponseT> implements Callable<ChunkUploadResponse<Re
       String uploadUrl,
       ChunkUploadRequest request,
       ApiCallContext callContext,
-      ResumableUploadCommand command) {
+      ResumableUploadCommand command,
+      UploadProgressTracker progressTracker) {
     this(
         uploadChunkCallable,
         queryStatusCallable,
@@ -99,6 +101,7 @@ class ChunkAttemptCallable<ResponseT> implements Callable<ChunkUploadResponse<Re
         request,
         callContext,
         command,
+        progressTracker,
         Long.MAX_VALUE,
         NanoClock.getDefaultClock());
   }
@@ -111,6 +114,7 @@ class ChunkAttemptCallable<ResponseT> implements Callable<ChunkUploadResponse<Re
       ChunkUploadRequest request,
       ApiCallContext callContext,
       ResumableUploadCommand command,
+      UploadProgressTracker progressTracker,
       long deadlineNanos,
       ApiClock clock) {
     this.uploadChunkCallable =
@@ -122,6 +126,7 @@ class ChunkAttemptCallable<ResponseT> implements Callable<ChunkUploadResponse<Re
     this.currentRequest = checkNotNull(request, "request must not be null");
     this.originalCallContext = checkNotNull(callContext, "callContext must not be null");
     this.currentCommand = checkNotNull(command, "command must not be null");
+    this.progressTracker = checkNotNull(progressTracker, "progressTracker must not be null");
     this.deadlineNanos = deadlineNanos;
     this.clock = checkNotNull(clock, "clock must not be null");
   }
@@ -164,6 +169,8 @@ private void prepareAttempt(
       ApiCallContext attemptContext,
       RetryingFuture<ChunkUploadResponse<ResponseT>> currentRetryingFuture) {
     this.currentCommand = ResumableUploadCommand.QUERY;
+    progressTracker.onRecovering(lastFailure);
+
     // Per GAX-R7: query uses sensible unary defaults trimmed to the remaining global deadline.
     long remainingNanos =
         deadlineNanos == Long.MAX_VALUE
@@ -259,6 +266,7 @@ private void handleQuerySuccess(
     // Normal path: realign buffer to committedOffset, compact and top up.
     try {
       buffer.realignTo(committedOffset);
+      progressTracker.onOffsetReceived(committedOffset);
     } catch (Throwable e) {
       failAttempt(attemptFuture, e);
       return;
 
@@ -87,6 +87,7 @@ final class ResumableUploadChunkCoordinator<ResponseT> {
   private final long deadlineNanos;
   private final ApiCallContext callContext;
   private final ClientContext clientContext;
+  private final UploadProgressTracker progressTracker;
   private final SettableApiFuture<ResponseT> result = SettableApiFuture.create();
   private volatile @Nullable ApiFuture<?> currentChunkFuture;
 
@@ -98,7 +99,8 @@ final class ResumableUploadChunkCoordinator<ResponseT> {
       ResumableUploadCallSettings settings,
       long deadlineNanos,
       ApiCallContext callContext,
-      ClientContext clientContext) {
+      ClientContext clientContext,
+      UploadProgressTracker progressTracker) {
     this.uploadChunkCallable =
         checkNotNull(uploadChunkCallable, "uploadChunkCallable must not be null");
     this.queryStatusCallable =
@@ -109,6 +111,7 @@ final class ResumableUploadChunkCoordinator<ResponseT> {
     this.deadlineNanos = deadlineNanos;
     this.callContext = checkNotNull(callContext, "callContext must not be null");
     this.clientContext = checkNotNull(clientContext, "clientContext must not be null");
+    this.progressTracker = checkNotNull(progressTracker, "progressTracker must not be null");
     this.buffer = new RewindableStreamBuffer(payload, settings.getChunkSize(), uploadUrl);
   }
 
@@ -200,6 +203,7 @@ private void transmitSingleChunk(long currentOffset) {
             chunkRequest,
             chunkCallContext,
             command,
+            progressTracker,
             deadlineNanos,
             clientContext.getClock());
 
@@ -224,6 +228,7 @@ public void onSuccess(ChunkUploadResponse<ResponseT> response) {
               return;
             }
             long nextOffset = buffer.getBufferBaseOffset() + buffer.getPayloadLength();
+            progressTracker.onChunkUploaded(nextOffset);
             if (response.getUploadStatus() == ResumableUploadStatus.FINAL) {
               result.set(response.getResponse());
             } else if (buffer.isFinal()) {
 
@@ -31,6 +31,7 @@
 
 import com.google.api.core.ApiFuture;
 import com.google.api.core.BetaApi;
+import java.util.concurrent.Executor;
 import org.jspecify.annotations.NullMarked;
 import org.jspecify.annotations.Nullable;
 
@@ -48,4 +49,18 @@ public interface ResumableUploadFuture<ResponseT> extends ApiFuture<ResponseT> {
 
   /** Returns the upload session URL, or {@code null} if session initiation is in progress. */
   @Nullable String getUploadSessionUrl();
+
+  /**
+   * Registers a listener to receive progress and state transition notifications for this upload.
+   *
+   * <p>A snapshot of the current upload status is dispatched to the listener immediately upon
+   * subscription on the provided executor. Subsequent status updates are delivered in order.
+   *
+   * @param listener callback listener to receive progress notifications
+   * @param executor executor on which the listener callbacks are dispatched
+   */
+  void addProgressListener(ResumableUploadProgressListener listener, Executor executor);
+
+  /** Returns the current progress snapshot of the upload session. */
+  ResumableUploadProgress getStatus();
 }
@@ -77,6 +77,7 @@ final class ResumableUploadFutureImpl<ResponseT> implements ResumableUploadFutur
   private final ApiCallContext callContext;
   private final ClientContext clientContext;
   private final ScheduledExecutorService executor;
+  private final UploadProgressTracker progressTracker = new UploadProgressTracker();
   private final SettableApiFuture<ResponseT> resultFuture = SettableApiFuture.create();
 
   private volatile @Nullable String uploadSessionUrl;
@@ -151,6 +152,7 @@ public void onSuccess(ResumableUploadSession session) {
               return;
             }
             uploadSessionUrl = session.getUploadUrl();
+            progressTracker.onStarted(uploadSessionUrl);
             ResumableUploadChunkCoordinator<ResponseT> coordinator =
                 new ResumableUploadChunkCoordinator<>(
                     uploadChunkCallable,
@@ -160,7 +162,8 @@ public void onSuccess(ResumableUploadSession session) {
                     settings,
                     deadlineNanos,
                     callContext,
-                    clientContext);
+                    clientContext,
+                    progressTracker);
             ApiFuture<ResponseT> uploadFuture;
             try {
               uploadFuture = coordinator.start();
@@ -224,6 +227,7 @@ private void succeed(@Nullable ResponseT result) {
     if (timeout != null) {
       timeout.cancel(false);
     }
+    progressTracker.onFinalized(progressTracker.getStatus().getBytesUploaded());
     closePayload();
     resultFuture.set(result);
   }
@@ -243,6 +247,7 @@ private void fail(Throwable t) {
     if (inFlight != null) {
       inFlight.cancel(true);
     }
+    progressTracker.onFailed(t, uploadSessionUrl);
     closePayload();
     resultFuture.setException(t);
   }
@@ -260,6 +265,16 @@ private void closePayload() {
     return uploadSessionUrl;
   }
 
+  @Override
+  public void addProgressListener(ResumableUploadProgressListener listener, Executor executor) {
+    progressTracker.addListener(listener, executor);
+  }
+
+  @Override
+  public ResumableUploadProgress getStatus() {
+    return progressTracker.getStatus();
+  }
+
   @Override
   public void addListener(Runnable listener, Executor executor) {
     resultFuture.addListener(listener, executor);
@@ -283,6 +298,7 @@ public boolean cancel(boolean mayInterruptIfRunning) {
     if (inFlight != null) {
       inFlight.cancel(mayInterruptIfRunning);
     }
+    progressTracker.onFailed(new CancellationException("Upload was cancelled"), uploadSessionUrl);
     closePayload();
     return cancelled;
   }
 
@@ -131,7 +131,8 @@ void call_successfulChunk_setsAttemptFuture() throws Exception {
             "https://upload.url/test",
             request,
             callContext,
-            ResumableUploadCommand.UPLOAD);
+            ResumableUploadCommand.UPLOAD,
+            new UploadProgressTracker());
 
     callable.setRetryingFuture(mockExternalFuture);
     ChunkUploadResponse<String> callResult = callable.call();
@@ -179,7 +180,8 @@ void call_returnsWithoutBlocking_andPropagatesCancellation() throws Exception {
             "https://upload.url/test",
             request,
             callContext,
-            ResumableUploadCommand.UPLOAD);
+            ResumableUploadCommand.UPLOAD,
+            new UploadProgressTracker());
 
     List<Runnable> listeners = new ArrayList<>();
     doAnswer(
@@ -253,7 +255,8 @@ void call_perAttemptDeadline_appliesRpcTimeoutToCallContext() throws Exception {
             "https://upload.url/test",
             request,
             callContext,
-            ResumableUploadCommand.UPLOAD);
+            ResumableUploadCommand.UPLOAD,
+            new UploadProgressTracker());
 
     callable.setRetryingFuture(mockExternalFuture);
     callable.call();
@@ -294,7 +297,8 @@ void call_nonBlockingExecution_callingThreadMakesImmediateProgress() throws Exce
             "https://upload.url/test",
             request,
             callContext,
-            ResumableUploadCommand.UPLOAD);
+            ResumableUploadCommand.UPLOAD,
+            new UploadProgressTracker());
 
     callable.setRetryingFuture(mockExternalFuture);