fix(retry): require explicit replay-safe opt-in for connection-reset retries

src/lib/upstream-retry.ts

@@ -2,10 +2,9 @@
  * Retry guard for upstream fetches that die on stale pooled keep-alive sockets.
  *
  * chatgpt.com (Cloudflare) closes idle keep-alive connections server-side; Bun's fetch pool
- * reuses the half-closed socket and the request write fails with ECONNRESET before any
- * response bytes arrive. Retrying on a fresh connection is safe for our replayable
- * (string-body) upstream requests, because fetch() rejects only before response headers —
- * a caught error here means no response was ever received.
+ * reuses the half-closed socket and the request write can fail with ECONNRESET before any
+ * response bytes arrive. A rejection before response headers does not prove that the origin
+ * did not process the request, so reset retries require an explicit replay-safety opt-in.
  *
  * Deliberately narrow: timeouts, aborts, ECONNREFUSED/DNS/TLS failures, and HTTP error
  * statuses (returned as Response, never thrown) are NOT retried. Mid-stream SSE resets are
@@ -238,6 +237,8 @@ export interface ResetRetryOptions {
   abortSignal?: AbortSignal;
   /** Short host/path label for the retry warn log (no secrets/query strings). */
   label?: string;
+  /** Opt in only when replaying the operation cannot duplicate upstream side effects. */
+  replaySafe?: boolean;
   attempts?: number;
 }
 
@@ -304,16 +305,20 @@ export function applyUpstreamRecoveryInit<T extends RequestInit>(
 }
 
 /**
- * Run `doFetch`, retrying only connection-reset-shaped rejections (see
- * isConnectionResetError) with jittered backoff. The caller's thunk must be replay-safe
- * (string body); every retry is logged so persistent resets stay visible.
+ * Run `doFetch`, retrying connection-reset-shaped rejections (see
+ * isConnectionResetError) with jittered backoff only when the caller explicitly proves the
+ * operation replay-safe. String bodies are mechanically reusable, but that alone does not
+ * make a model request idempotent: the origin may process it before closing the connection.
+ * Every retry is logged so persistent resets stay visible.
  */
 export async function fetchWithResetRetry(
   doFetch: ReplayableFetch,
   opts: ResetRetryOptions = {},
   firstRecovery?: UpstreamSendRecovery,
 ): Promise<Response> {
-  const attempts = Math.max(1, opts.attempts ?? RESET_RETRY_MAX_ATTEMPTS);
+  const attempts = opts.replaySafe
+    ? Math.max(1, opts.attempts ?? RESET_RETRY_MAX_ATTEMPTS)
+    : 1;
   let lastError: unknown;
   let sawReset = false;
   for (let attempt = 0; attempt < attempts; attempt++) {

tests/issue-914-transport-attribution.test.ts

@@ -146,7 +146,7 @@ describe("issue #914 — pre-connection failures never touch account health", ()
     const err = await fetchWithResetRetry(async recovery => {
       if (!recovery) throw coded("reset", "ECONNRESET");
       throw rejection;
-    }).catch((e: unknown) => e);
+    }, { replaySafe: true }).catch((e: unknown) => e);
     expect(classifyTransportFailureKind(err)).toBe("connect_error");
   });
 

tests/upstream-retry.test.ts

@@ -147,7 +147,7 @@ describe("fetchWithResetRetry", () => {
   test("retries a Bun-shaped reset and returns the second attempt's response", async () => {
     silenceWarn();
     const mock = mockDoFetch([bunResetError(), new Response("ok", { status: 200 })]);
-    const res = await fetchWithResetRetry(mock.doFetch, { label: "test" });
+    const res = await fetchWithResetRetry(mock.doFetch, { label: "test", replaySafe: true });
     expect(res.status).toBe(200);
     expect(await res.text()).toBe("ok");
     expect(mock.calls).toHaveLength(2);
@@ -160,7 +160,7 @@ describe("fetchWithResetRetry", () => {
       new Error("The socket connection was closed unexpectedly."),
       new Response("ok", { status: 200 }),
     ]);
-    const res = await fetchWithResetRetry(mock.doFetch);
+    const res = await fetchWithResetRetry(mock.doFetch, { replaySafe: true });
     expect(res.status).toBe(200);
     expect(mock.calls).toHaveLength(2);
   });
@@ -179,6 +179,12 @@ describe("fetchWithResetRetry", () => {
     expect(mock.calls).toHaveLength(1);
   });
 
+  test("does not replay a reset unless the operation is explicitly replay-safe", async () => {
+    const mock = mockDoFetch([bunResetError(), new Response("ok", { status: 200 })]);
+    await expect(fetchWithResetRetry(mock.doFetch)).rejects.toThrow("socket connection was closed unexpectedly");
+    expect(mock.calls).toHaveLength(1);
+  });
+
   test("passes HTTP error responses through without retrying", async () => {
     const mock = mockDoFetch([new Response("upstream boom", { status: 502 })]);
     const res = await fetchWithResetRetry(mock.doFetch);
@@ -189,7 +195,7 @@ describe("fetchWithResetRetry", () => {
   test("gives up after max attempts and rethrows the last reset error", async () => {
     silenceWarn();
     const mock = mockDoFetch([bunResetError(), bunResetError(), bunResetError(), bunResetError()]);
-    await expect(fetchWithResetRetry(mock.doFetch)).rejects.toThrow("socket connection was closed unexpectedly");
+    await expect(fetchWithResetRetry(mock.doFetch, { replaySafe: true })).rejects.toThrow("socket connection was closed unexpectedly");
     expect(mock.calls).toHaveLength(3);
     expect(warnSpies[0]).toHaveBeenCalledTimes(2);
   });
@@ -206,7 +212,7 @@ describe("fetchWithResetRetry", () => {
     silenceWarn();
     const ac = new AbortController();
     const mock = mockDoFetch([bunResetError(), new Response("ok", { status: 200 })]);
-    const pending = fetchWithResetRetry(mock.doFetch, { abortSignal: ac.signal });
+    const pending = fetchWithResetRetry(mock.doFetch, { abortSignal: ac.signal, replaySafe: true });
     // First attempt rejects with a reset synchronously-ish; abort lands mid-backoff.
     setTimeout(() => ac.abort(new DOMException("client closed", "AbortError")), 10);
     await expect(pending).rejects.toThrow("client closed");