diff --git a/CHANGELOG.md b/CHANGELOG.md
index 873016a54..6863023aa 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -4,6 +4,13 @@ All notable changes to this extension will be documented in this file.
 
 ## Unreleased
 
+### Fixed
+
+- Flink statement results no longer stop loading when Confluent Cloud returns a temporary error
+  right after a statement is submitted. The Results Viewer now retries before giving up, waiting as
+  long as Confluent Cloud asks when it sends a rate-limit delay, instead of showing "Failed to load
+  results."
+
 ## 2.3.1
 
 ### Fixed
diff --git a/src/errors.test.ts b/src/errors.test.ts
index fe14c5d70..a6ba83230 100644
--- a/src/errors.test.ts
+++ b/src/errors.test.ts
@@ -7,6 +7,7 @@ import {
   getNestedErrorChain,
   hasErrorCause,
   isResponseErrorWithStatus,
+  isTransientResponseError,
   logError,
 } from "./errors";
 import { Logger } from "./logging";
@@ -169,6 +170,27 @@ describe("errors.ts isResponseErrorWithStatus()", () => {
   });
 });
 
+describe("errors.ts isTransientResponseError()", () => {
+  it("should return false for not-a-response-error", () => {
+    const error = new Error("test");
+    assert.strictEqual(isTransientResponseError(error), false);
+  });
+
+  for (const status of [429, 500, 502, 503, 504]) {
+    it(`should return true for a ${status} response error`, () => {
+      const error = createResponseError(status, "Transient", "test");
+      assert.strictEqual(isTransientResponseError(error), true);
+    });
+  }
+
+  for (const status of [400, 401, 403, 404, 409]) {
+    it(`should return false for a ${status} response error`, () => {
+      const error = createResponseError(status, "Client Error", "test");
+      assert.strictEqual(isTransientResponseError(error), false);
+    });
+  }
+});
+
 describe("errors.ts extractResponseBody()", () => {
   it("should return the response body as JSON if it is valid JSON", async () => {
     const embeddedObject = { message: "test" };
diff --git a/src/errors.ts b/src/errors.ts
index 04b260ac6..6345ac0be 100644
--- a/src/errors.ts
+++ b/src/errors.ts
@@ -55,6 +55,14 @@ export function isResponseErrorWithStatus(
   return isResponseError(error) && error.response.status === statusCode;
 }
 
+/** HTTP statuses where repeating the same request after a short delay may succeed. */
+const TRANSIENT_RESPONSE_STATUSES = new Set([429, 500, 502, 503, 504]);
+
+/** Was this a response error whose status suggests the request is worth retrying? */
+export function isTransientResponseError(error: unknown): error is AnyResponseError {
+  return isResponseError(error) && TRANSIENT_RESPONSE_STATUSES.has(error.response.status);
+}
+
 /**
  * If error is a response error, try to decode its response body
  * from JSON and return the resulting object.
diff --git a/src/flinkSql/flinkStatementResultsManager.test.ts b/src/flinkSql/flinkStatementResultsManager.test.ts
index 71444310d..f137d4bcd 100644
--- a/src/flinkSql/flinkStatementResultsManager.test.ts
+++ b/src/flinkSql/flinkStatementResultsManager.test.ts
@@ -5,7 +5,11 @@ import type { FlinkStatementResultsManagerTestContext } from "../../tests/create
 import { createTestResultsManagerContext } from "../../tests/createResultsManager";
 import { eventually } from "../../tests/eventually";
 import { loadFixtureFromFile } from "../../tests/fixtures/utils";
-import { createResponseError } from "../../tests/unit/testUtils";
+import {
+  createResponseError,
+  createSingleUseResponseError,
+  ResponseErrorSource,
+} from "../../tests/unit/testUtils";
 import type { GetSqlv1StatementResult200Response } from "../clients/flinkSql";
 import {
   GetSqlv1StatementResult200ResponseApiVersionEnum,
@@ -18,6 +22,15 @@ import type {
   FlinkStatementResultsViewModel,
   ResultsViewerStorageState,
 } from "../webview/flink-statement-results";
+import { transientBackoffWindow } from "./flinkStatementResultsManager";
+
+/** A successful results response carrying no rows. */
+const EMPTY_RESULTS_RESPONSE: GetSqlv1StatementResult200Response = {
+  api_version: GetSqlv1StatementResult200ResponseApiVersionEnum.SqlV1,
+  kind: GetSqlv1StatementResult200ResponseKindEnum.StatementResult,
+  metadata: {},
+  results: { data: [] },
+};
 
 function createMockStatement(): FlinkStatement {
   const fakeFlinkStatement = loadFixtureFromFile(
@@ -401,14 +414,9 @@ describe("FlinkStatementResultsViewModel and FlinkStatementResultsManager", () =
       ctx.flinkSqlStatementResultsApi.getSqlv1StatementResult
         .onSecondCall()
         .rejects(createResponseError(409, "Conflict", "{}"));
-      ctx.flinkSqlStatementResultsApi.getSqlv1StatementResult.onThirdCall().resolves({
-        api_version: GetSqlv1StatementResult200ResponseApiVersionEnum.SqlV1,
-        kind: GetSqlv1StatementResult200ResponseKindEnum.StatementResult,
-        metadata: {},
-        results: {
-          data: [],
-        },
-      });
+      ctx.flinkSqlStatementResultsApi.getSqlv1StatementResult
+        .onThirdCall()
+        .resolves(EMPTY_RESULTS_RESPONSE);
 
       // Trigger a fetch
       const fetchPromise = ctx.manager.fetchResults();
@@ -442,9 +450,8 @@ describe("FlinkStatementResultsViewModel and FlinkStatementResultsManager", () =
       assert.ok(ctx.manager["_latestError"]());
     });
 
-    it("should not retry on non-409 errors during fetch", async () => {
-      // Mock the getSqlv1StatementResult to fail with 500
-      const responseError = createResponseError(500, "Internal Server Error", "{}");
+    it("should not retry on errors that are neither 409 nor transient during fetch", async () => {
+      const responseError = createResponseError(403, "Forbidden", "{}");
       ctx.flinkSqlStatementResultsApi.getSqlv1StatementResult.rejects(responseError);
 
       // Trigger a fetch
@@ -455,11 +462,140 @@ describe("FlinkStatementResultsViewModel and FlinkStatementResultsManager", () =
 
       await fetchPromise;
 
-      assert.equal(ctx.flinkSqlStatementResultsApi.getSqlv1StatementResult.callCount, 1);
+      sinon.assert.calledOnce(ctx.flinkSqlStatementResultsApi.getSqlv1StatementResult);
       // Verify error state is set
       assert.ok(ctx.manager["_latestError"]());
     });
 
+    it("should retry get statement results on transient errors", async () => {
+      // CCloud briefly can't resolve a just-created statement, answering 429 or 5xx before the
+      // results endpoint starts working
+      ctx.flinkSqlStatementResultsApi.getSqlv1StatementResult
+        .onFirstCall()
+        .rejects(createResponseError(429, "Too Many Requests", "{}"));
+      ctx.flinkSqlStatementResultsApi.getSqlv1StatementResult
+        .onSecondCall()
+        .rejects(createResponseError(500, "Internal Server Error", "{}"));
+      ctx.flinkSqlStatementResultsApi.getSqlv1StatementResult
+        .onThirdCall()
+        .resolves(EMPTY_RESULTS_RESPONSE);
+
+      const fetchPromise = ctx.manager.fetchResults();
+
+      // backoff doubles from 500ms and is jittered, so tick past the two maximums
+      await clock.tickAsync(500 + 1000);
+
+      await fetchPromise;
+
+      sinon.assert.calledThrice(ctx.flinkSqlStatementResultsApi.getSqlv1StatementResult);
+      assert.equal(ctx.manager["_latestError"](), null);
+    });
+
+    it("should wait for the server's Retry-After rather than the exponential curve", async () => {
+      ctx.flinkSqlStatementResultsApi.getSqlv1StatementResult.onFirstCall().rejects(
+        createResponseError(429, "Too Many Requests", "{}", ResponseErrorSource.Sidecar, {
+          "retry-after": "2",
+        }),
+      );
+      ctx.flinkSqlStatementResultsApi.getSqlv1StatementResult
+        .onSecondCall()
+        .resolves(EMPTY_RESULTS_RESPONSE);
+
+      const fetchPromise = ctx.manager.fetchResults();
+
+      // the exponential curve would have retried by now, but the server asked for 2s
+      await clock.tickAsync(1000);
+      sinon.assert.calledOnce(ctx.flinkSqlStatementResultsApi.getSqlv1StatementResult);
+
+      // 2s as requested, plus up to one base delay of jitter on top
+      await clock.tickAsync(1500);
+      await fetchPromise;
+
+      sinon.assert.calledTwice(ctx.flinkSqlStatementResultsApi.getSqlv1StatementResult);
+      assert.equal(ctx.manager["_latestError"](), null);
+    });
+
+    it("should complete the stream after exhausting transient retries", async () => {
+      ctx.flinkSqlStatementResultsApi.getSqlv1StatementResult.rejects(
+        createResponseError(500, "Internal Server Error", "{}"),
+      );
+
+      const fetchPromise = ctx.manager.fetchResults();
+
+      // 4 transient retries at up to 500/1000/2000/4000ms
+      await clock.tickAsync(7500);
+
+      await fetchPromise;
+
+      // 1 initial attempt + MAX_TRANSIENT_RETRIES
+      sinon.assert.callCount(ctx.flinkSqlStatementResultsApi.getSqlv1StatementResult, 5);
+      assert.ok(ctx.manager["_latestError"]());
+      assert.equal(ctx.manager["_state"](), "completed");
+    });
+
+    it("should stop waiting out a transient backoff when the manager is disposed", async () => {
+      const stub = ctx.flinkSqlStatementResultsApi.getSqlv1StatementResult;
+      stub.rejects(createResponseError(500, "Internal Server Error", "{}"));
+
+      const fetchPromise = ctx.manager.fetchResults();
+      // let the first attempt fail and settle into its backoff
+      await clock.tickAsync(1);
+      sinon.assert.calledOnce(stub);
+
+      ctx.manager.dispose();
+
+      // settles without the clock ever reaching the end of that backoff, which is the point: a real
+      // aborted signal would also end the loop, but only after the full wait
+      await fetchPromise;
+    });
+
+    it("should not report a disposal-interrupted fetch as an error", async () => {
+      const stub = ctx.flinkSqlStatementResultsApi.getSqlv1StatementResult;
+      stub.rejects(createResponseError(500, "Internal Server Error", "{}"));
+
+      const fetchPromise = ctx.manager.fetchResults();
+      await clock.tickAsync(1);
+      ctx.manager.dispose();
+      await fetchPromise;
+
+      // aborting rethrows the 500 that started the retry; surfacing it would toast the user for
+      // closing the results pane
+      assert.equal(ctx.manager["_latestError"](), null);
+    });
+
+    it("should not let transient retries eat into the 409 budget", async () => {
+      // 4 transient retries first, exhausting that budget, then nothing but 409s
+      const stub = ctx.flinkSqlStatementResultsApi.getSqlv1StatementResult;
+      for (let i = 0; i < 4; i++) {
+        stub.onCall(i).rejects(createResponseError(500, "Internal Server Error", "{}"));
+      }
+      stub.rejects(createResponseError(409, "Conflict", "{}"));
+
+      const fetchPromise = ctx.manager.fetchResults();
+
+      // 7500ms covers the transient backoffs, then the 409 waits with a little margin
+      await clock.tickAsync(7500 + 61 * 500);
+
+      await fetchPromise;
+
+      // 4 transient calls + the 409s' full 60-attempt budget, untouched by them
+      sinon.assert.callCount(stub, 64);
+    });
+
+    it("should leave the error response body readable for logging", async () => {
+      // a real single-use Response, so reading the body without cloning would be observable
+      const responseError = createSingleUseResponseError(
+        400,
+        "Bad Request",
+        '{"errors":[{"code":"cr_failed_get_stmt_name"}]}',
+      );
+      ctx.flinkSqlStatementResultsApi.getSqlv1StatementResult.rejects(responseError);
+
+      await ctx.manager.fetchResults();
+
+      assert.strictEqual(responseError.response.bodyUsed, false);
+    });
+
     it("should only allow one instance of fetchResults to run at a time", async () => {
       // Create a promise that we can resolve manually to simulate a slow API call
       let resolveRequest: (value: GetSqlv1StatementResult200Response) => void;
@@ -486,12 +622,7 @@ describe("FlinkStatementResultsViewModel and FlinkStatementResultsManager", () =
       assert.equal(ctx.flinkSqlStatementResultsApi.getSqlv1StatementResult.callCount, 1);
 
       // Resolve the API call
-      resolveRequest!({
-        api_version: GetSqlv1StatementResult200ResponseApiVersionEnum.SqlV1,
-        kind: GetSqlv1StatementResult200ResponseKindEnum.StatementResult,
-        metadata: {},
-        results: { data: [] },
-      });
+      resolveRequest!(EMPTY_RESULTS_RESPONSE);
 
       // Wait for all calls to complete
       await Promise.all(fetchPromises);
@@ -768,3 +899,65 @@ describe("FlinkStatementResultsViewModel only", () => {
     }
   });
 });
+
+describe("flinkStatementResultsManager.ts transientBackoffWindow()", () => {
+  function responseWithHeaders(headers: Record<string, string>): Response {
+    return new Response("{}", { status: 429, headers });
+  }
+
+  it("should honor a Retry-After the server sends, never shortening it", () => {
+    const { minMs, maxMs } = transientBackoffWindow(responseWithHeaders({ "retry-after": "2" }), 0);
+
+    assert.equal(minMs, 2000);
+    assert.ok(maxMs > minMs, "jitter should only extend a server-requested delay");
+  });
+
+  it("should cap an outsized Retry-After", () => {
+    const { minMs } = transientBackoffWindow(responseWithHeaders({ "retry-after": "600" }), 0);
+
+    assert.equal(minMs, 8000);
+  });
+
+  it("should fall back to X-RateLimit-Reset when a 429 omits Retry-After", () => {
+    const { minMs } = transientBackoffWindow(
+      responseWithHeaders({ "x-ratelimit-limit": "5", "x-ratelimit-reset": "3" }),
+      0,
+    );
+
+    assert.equal(minMs, 3000);
+  });
+
+  it("should prefer Retry-After over X-RateLimit-Reset when both are present", () => {
+    const { minMs } = transientBackoffWindow(
+      responseWithHeaders({ "retry-after": "1", "x-ratelimit-reset": "3" }),
+      0,
+    );
+
+    assert.equal(minMs, 1000);
+  });
+
+  it("should fall back to an exponential curve without either header", () => {
+    const windows = [0, 1, 2, 3].map((attempt) =>
+      transientBackoffWindow(responseWithHeaders({}), attempt),
+    );
+
+    assert.deepEqual(
+      windows.map((w) => w.maxMs),
+      [500, 1000, 2000, 4000],
+    );
+    assert.deepEqual(
+      windows.map((w) => w.minMs),
+      [250, 500, 1000, 2000],
+    );
+  });
+
+  it("should fall back to the curve for a non-numeric Retry-After", () => {
+    // RFC 7231 also permits an HTTP-date, which we don't parse
+    const { maxMs } = transientBackoffWindow(
+      responseWithHeaders({ "retry-after": "Wed, 21 Oct 2015 07:28:00 GMT" }),
+      0,
+    );
+
+    assert.equal(maxMs, 500);
+  });
+});
diff --git a/src/flinkSql/flinkStatementResultsManager.ts b/src/flinkSql/flinkStatementResultsManager.ts
index b0ac9c685..7cee85ce6 100644
--- a/src/flinkSql/flinkStatementResultsManager.ts
+++ b/src/flinkSql/flinkStatementResultsManager.ts
@@ -9,7 +9,13 @@ import type {
 } from "../clients/flinkSql";
 import { FetchError } from "../clients/flinkSql";
 import { showJsonPreview } from "../documentProviders/message";
-import { isResponseError, isResponseErrorWithStatus, logError } from "../errors";
+import {
+  extractResponseBody,
+  isResponseError,
+  isResponseErrorWithStatus,
+  isTransientResponseError,
+  logError,
+} from "../errors";
 import { CCloudResourceLoader } from "../loaders/ccloudResourceLoader";
 import { Logger } from "../logging";
 import type { FlinkStatement } from "../models/flinkStatement";
@@ -19,13 +25,28 @@ import type { ViewMode } from "./flinkStatementResultColumns";
 import type { StatementResultsRow } from "./flinkStatementResults";
 import { parseResults } from "./flinkStatementResults";
 import { extractPageToken } from "./utils";
+import { pauseWithJitter } from "../utils/timing";
 import type { SqlV1StatementWarning } from "../clients/flinkSql";
 
 const logger = new Logger("flink-statement-results");
 
+/** Attempts allowed while the statement's results are still being prepared (HTTP 409). */
+const MAX_CONFLICT_RETRIES = 60;
+/** Constant delay between 409 attempts. */
+const CONFLICT_BACKOFF_MS = 500;
+/** Retries allowed per transient response, on top of the initial attempt. */
+const MAX_TRANSIENT_RETRIES = 4;
+/** First transient backoff delay; doubles each retry, so 500/1000/2000/4000ms. */
+const TRANSIENT_BASE_BACKOFF_MS = 500;
+/** Ceiling on any transient backoff, so an outsized `Retry-After` can't stall the viewer. */
+const MAX_TRANSIENT_BACKOFF_MS = 8_000;
+
 export type ResultCount = { total: number; filter: number | null };
 export type StreamState = "running" | "completed";
 
+/** Retry attempts already spent, per class of failure. */
+type RetryBudget = { conflict: number; transient: number };
+
 export type MessageType =
   | "GetResults"
   | "GetResultsCount"
@@ -229,19 +250,23 @@ export class FlinkStatementResultsManager {
         const priorRawResults = this._rawResults();
         const pageToken = extractPageToken(this._latestResult()?.metadata?.next);
 
-        const response = await this.retry(async () => {
-          return await this._flinkStatementResultsSqlApi.getSqlv1StatementResult(
-            {
-              environment_id: this.statement.environmentId,
-              organization_id: this.statement.organizationId,
-              name: this.statement.name,
-              page_token: pageToken,
-            },
-            {
-              signal: this._getResultsAbortController.signal,
-            },
-          );
-        }, "fetch statement results");
+        const response = await this.retry(
+          async () => {
+            return await this._flinkStatementResultsSqlApi.getSqlv1StatementResult(
+              {
+                environment_id: this.statement.environmentId,
+                organization_id: this.statement.organizationId,
+                name: this.statement.name,
+                page_token: pageToken,
+              },
+              {
+                signal: this._getResultsAbortController.signal,
+              },
+            );
+          },
+          "fetch statement results",
+          true,
+        );
 
         const resultsData: SqlV1StatementResultResults = response.results ?? {};
 
@@ -269,13 +294,14 @@ export class FlinkStatementResultsManager {
           this.notifyUI();
         });
       } catch (error) {
-        if (error instanceof FetchError && error?.cause?.name === "AbortError") {
+        if (this.wasFetchAborted(error)) {
           logger.info("Statement results fetch was aborted");
           return;
         }
 
         if (isResponseError(error)) {
-          const payload = await error.response.json();
+          // clone before reading, so logError() below can still read the body for Sentry
+          const payload = await extractResponseBody(error);
           if (!payload?.aborted) {
             const status = error.response.status;
             shouldComplete = status >= 400;
@@ -360,29 +386,28 @@ export class FlinkStatementResultsManager {
   }
 
   /**
-   * Retries {@link maxRetries} times with a constant backoff delay of
-   * {@link backoffMs}. Nothing fancy.
+   * Retries a 409 conflict up to {@linkcode MAX_CONFLICT_RETRIES} times with a constant backoff
+   * delay of {@linkcode CONFLICT_BACKOFF_MS}. Nothing fancy.
+   *
+   * When {@linkcode retryTransient} is set, statuses that may clear on their own (429 and 5xx) also
+   * get up to {@linkcode MAX_TRANSIENT_RETRIES} attempts on a separate backoff budget, delegated to
+   * {@linkcode transientBackoffWindow}. Only safe for idempotent operations.
    */
   private async retry<T>(
     operation: () => Promise<T>,
     operationName: string,
-    maxRetries: number = 60,
-    backoffMs: number = 500,
+    retryTransient: boolean = false,
   ): Promise<T> {
     let lastErr: Error | undefined;
-    for (let attempt = 0; attempt < maxRetries; attempt++) {
+    const spent: RetryBudget = { conflict: 0, transient: 0 };
+    // each budget is bounded by its own counter, so this ceiling should never be what stops us;
+    // it is here only so the loop is self-evidently finite
+    for (let i = 0; i <= MAX_CONFLICT_RETRIES + MAX_TRANSIENT_RETRIES; i++) {
       try {
         return await operation();
       } catch (err) {
         lastErr = err as Error;
-        if (isResponseErrorWithStatus(err, 409)) {
-          if (attempt < maxRetries - 1) {
-            logger.debug(
-              `Retrying ${operationName} after 409 conflict. Attempt ${attempt + 1}/${maxRetries}. Waiting ${backoffMs}ms`,
-            );
-            await new Promise((resolve) => setTimeout(resolve, backoffMs));
-          }
-        } else {
+        if (!(await this.backOffBeforeRetry(err, operationName, retryTransient, spent))) {
           break;
         }
       }
@@ -391,6 +416,84 @@ export class FlinkStatementResultsManager {
     throw lastErr;
   }
 
+  /**
+   * Did this failure come from the fetch being abandoned rather than from the server? Aborting
+   * mid-backoff rethrows whatever failure started the retry, so the signal has to be checked as
+   * well, or closing the results pane would report that failure to the user.
+   */
+  private wasFetchAborted(error: unknown): boolean {
+    return (
+      this._getResultsAbortController.signal.aborted ||
+      (error instanceof FetchError && error?.cause?.name === "AbortError")
+    );
+  }
+
+  /**
+   * Wait out a retryable failure, charging it to the matching budget in {@linkcode spent}. Returns
+   * false when the error isn't retryable, its budget is spent, or the fetch was aborted while
+   * waiting, all of which mean {@linkcode retry} should give up.
+   */
+  private async backOffBeforeRetry(
+    err: unknown,
+    operationName: string,
+    retryTransient: boolean,
+    spent: RetryBudget,
+  ): Promise<boolean> {
+    if (isResponseErrorWithStatus(err, 409)) {
+      if (spent.conflict >= MAX_CONFLICT_RETRIES - 1) {
+        return false;
+      }
+      spent.conflict++;
+      logger.debug(
+        `Retrying ${operationName} after 409 conflict. Attempt ${spent.conflict}/${MAX_CONFLICT_RETRIES}. Waiting ${CONFLICT_BACKOFF_MS}ms`,
+      );
+      await new Promise((resolve) => setTimeout(resolve, CONFLICT_BACKOFF_MS));
+      return true;
+    }
+
+    if (
+      retryTransient &&
+      isTransientResponseError(err) &&
+      spent.transient < MAX_TRANSIENT_RETRIES
+    ) {
+      const { minMs, maxMs } = transientBackoffWindow(err.response, spent.transient);
+      spent.transient++;
+      logger.debug(
+        `Retrying ${operationName} after status ${err.response.status}. Transient attempt ${spent.transient}/${MAX_TRANSIENT_RETRIES}`,
+      );
+      await this.pauseUnlessAborted(minMs, maxMs);
+      return !this._getResultsAbortController.signal.aborted;
+    }
+
+    return false;
+  }
+
+  /**
+   * Sleep for a jittered interval, returning as soon as the results fetch is aborted so a backoff
+   * can't outlive {@linkcode dispose}. Only the results fetch opts into transient retries, so this
+   * deliberately watches that controller; `stopStatement()` aborts it before its own retries and
+   * must not be cut short.
+   */
+  private async pauseUnlessAborted(minMs: number, maxMs: number): Promise<void> {
+    const { signal } = this._getResultsAbortController;
+    if (signal.aborted) {
+      return;
+    }
+
+    let stopWatchingAbort = () => {};
+    const abortedEarly = new Promise<void>((resolve) => {
+      const onAbort = () => resolve();
+      signal.addEventListener("abort", onAbort, { once: true });
+      stopWatchingAbort = () => signal.removeEventListener("abort", onAbort);
+    });
+
+    try {
+      await Promise.race([pauseWithJitter(minMs, maxMs), abortedEarly]);
+    } finally {
+      stopWatchingAbort();
+    }
+  }
+
   private async stopStatement(): Promise<void> {
     // Abort any in-flight GET results requests
     this._getResultsAbortController.abort();
@@ -539,3 +642,37 @@ export class FlinkStatementResultsManager {
     this._getResultsAbortController.abort();
   }
 }
+
+/**
+ * Pick the backoff window for a transient response, preferring a delay Confluent Cloud asked for
+ * over a locally-guessed exponential one.
+ *
+ * `Retry-After` is only sent once a rate limit is actually hit, so `X-RateLimit-Reset` (which rides
+ * along on every response) covers a 429 that omits it. A server-requested delay is only ever
+ * extended by jitter, never shortened, since retrying before the window resets just earns another
+ * 429. The exponential fallback covers 5xx responses, which carry neither header.
+ */
+export function transientBackoffWindow(
+  response: Response,
+  attempt: number,
+): { minMs: number; maxMs: number } {
+  const serverDelaySeconds =
+    relativeSeconds(response.headers?.get("retry-after")) ??
+    relativeSeconds(response.headers?.get("x-ratelimit-reset"));
+  if (serverDelaySeconds !== undefined) {
+    const minMs = Math.min(serverDelaySeconds * 1000, MAX_TRANSIENT_BACKOFF_MS);
+    return { minMs, maxMs: minMs + TRANSIENT_BASE_BACKOFF_MS };
+  }
+
+  const maxMs = Math.min(TRANSIENT_BASE_BACKOFF_MS * 2 ** attempt, MAX_TRANSIENT_BACKOFF_MS);
+  return { minMs: maxMs / 2, maxMs };
+}
+
+/**
+ * Read a header carrying a positive number of seconds, ignoring absent values and the HTTP-date
+ * form of `Retry-After` that RFC 7231 also permits.
+ */
+function relativeSeconds(header: string | null | undefined): number | undefined {
+  const seconds = Number(header);
+  return Number.isFinite(seconds) && seconds > 0 ? seconds : undefined;
+}
diff --git a/tests/unit/testUtils.ts b/tests/unit/testUtils.ts
index da5b453d3..fd2241b5f 100644
--- a/tests/unit/testUtils.ts
+++ b/tests/unit/testUtils.ts
@@ -121,6 +121,7 @@ export enum ResponseErrorSource {
  * @param statusText - HTTP status text
  * @param body - Response body
  * @param source - Which {@link ResponseErrorSource client source} ResponseError is returned, defaults to sidecar
+ * @param headers - Response headers, for code that reads things like `Retry-After`
  * @returns A ResponseError instance
  */
 export function createResponseError(
@@ -128,10 +129,12 @@ export function createResponseError(
   statusText: string,
   body: string,
   source: ResponseErrorSource = ResponseErrorSource.Sidecar,
+  headers: Record<string, string> = {},
 ): AnyResponseError {
   const response = {
     status,
     statusText,
+    headers: new Headers(headers),
     clone: () => ({
       text: () => Promise.resolve(body),
       json: () => Promise.resolve(JSON.parse(body)),
@@ -140,8 +143,28 @@ export function createResponseError(
     json: () => Promise.resolve(JSON.parse(body)),
   } as Response;
 
-  // any callers that end up using `isResponseError()` will need to know which client code subdir
-  // the error came from, so we need to return the correct subclass of ResponseError
+  return wrapResponseError(response, source);
+}
+
+/**
+ * Create a mock ResponseError backed by a real single-use {@link Response}, so a body consumed
+ * without `.clone()` is observable. {@linkcode createResponseError}'s stand-in is re-readable
+ * forever and can't catch that.
+ */
+export function createSingleUseResponseError(
+  status: number,
+  statusText: string,
+  body: string,
+  source: ResponseErrorSource = ResponseErrorSource.Sidecar,
+): AnyResponseError {
+  return wrapResponseError(new Response(body, { status, statusText }), source);
+}
+
+/**
+ * Wrap a response in the ResponseError subclass matching its client source, since callers relying
+ * on `isResponseError()` need the subdir the error came from.
+ */
+function wrapResponseError(response: Response, source: ResponseErrorSource): AnyResponseError {
   switch (source) {
     case ResponseErrorSource.Docker:
       return new DockerResponseError(response);