/**
 * 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 a request can fail before response headers arrive.
 * A pre-header rejection does not prove that the origin did not process the request.
 * Mechanically reusable bytes do not make a model POST idempotent: an ambiguous reset
 * becomes a terminal, non-replayable response unless the operation is explicitly safe.
 *
 * Deliberately narrow: timeouts, aborts, ECONNREFUSED/DNS/TLS failures, and HTTP error
 * statuses (returned as Response, never thrown) are NOT retried. A reset after the response
 * head is out of scope here, because the response has already resolved by then; the Responses
 * transport asks the same question at that stage through the shared resend gate.
 *
 * MUST stay a leaf module: imports nothing from server.ts or adapters (kiro-retry imports
 * the shared abort helpers from here).
 */
import { clearableDeadline } from "./abort";
import { isDefiniteContextOverflow } from "./context-overflow-verdict";
import { errorDestinationLabel, errorDestinationText, type ErrorDestinationLabel } from "./error-destination";
export { errorDestinationLabel, providerErrorDestinationLabel, errorDestinationText } from "./error-destination";
import { redactSecretString } from "./redact";

/**
 * Responses the origin may already be executing. RFC 9110 §9.2.2 forbids an intermediary
 * from automatically repeating a non-idempotent request; when a post-send transport (the
 * Codex WebSocket relay) settles a gateway status because the origin never acknowledged the
 * turn, the request body must not be sent again by this process — not by the transient-5xx
 * layer below, not by a pool account rotation, not by a combo hop. The status is returned
 * to the caller so the user agent can apply its own retry policy, exactly as it does on the
 * direct path. The WeakSet is the in-process marker; the structured error codes are the
 * marker that survives body re-wrapping (combo failure consumption re-parses the JSON).
 */
const nonReplayableResponses = new WeakSet<Response>();

export function markResponseNonReplayable(response: Response): void {
  nonReplayableResponses.add(response);
}

/**
 * A server-instructed wait longer than the retry budget: same-target 429 replay loops stop and
 * return the original response, while key rotation and ordinary error delivery are unaffected.
 */
const sameTargetRetryHolds = new WeakSet<Response>();

export function markSameTargetRetryHold(response: Response): void {
  sameTargetRetryHolds.add(response);
}

export function isSameTargetRetryHold(response: Response): boolean {
  return sameTargetRetryHolds.has(response);
}

export function isNonReplayableResponse(response: Response): boolean {
  return nonReplayableResponses.has(response);
}

/**
 * The narrower marker: responses this proxy synthesized as a replay refusal.
 *
 * {@link isNonReplayableResponse} answers "must not be sent again", which the WebSocket
 * post-send verdicts share. This one answers "the upstream never said this", and that is the
 * question a quota recorder or a `Retry-After` synthesizer has to ask. Both were written for
 * a status that only ever arrived from a provider, so a synthetic 429 reads to them as a
 * credential that rate-limited us and as a wait worth honouring -- one writes a cooldown
 * against a credential that refused nothing, the other instructs the client to send the turn
 * again. A marker rather than a body check, because it has to be answerable before the body
 * is read and cannot be spoofed by an upstream that happens to echo the code.
 */
const replayRefusalResponses = new WeakSet<Response>();

export function markReplayRefusalResponse(response: Response): void {
  replayRefusalResponses.add(response);
}

export function isReplayRefusalResponse(response: Response): boolean {
  return replayRefusalResponses.has(response);
}

/** Origin never produced a response event; the turn may still be executing. */
export const UPSTREAM_NO_RESPONSE_CODE = "upstream_no_response";
/** Transport closed after the send, before any response event. */
export const UPSTREAM_CLOSED_BEFORE_RESPONSE_CODE = "upstream_closed_before_response";
/**
 * This proxy refused to replay a pre-header fetch rejection.
 *
 * Distinct from {@link UPSTREAM_CLOSED_BEFORE_RESPONSE_CODE}, which the Codex WebSocket
 * transport settles as a 502 after the create frame was already sent. Both are ambiguous,
 * but only this one is a refusal this process made before any response existed, so it
 * follows the send-budget precedent and answers 429: the Codex client is configured with
 * `retry_429: false` and `retry_5xx: true` over four attempts, so a 5xx here multiplies
 * the duplicate send the refusal exists to prevent. See
 * structure/transports/responses.md#ambiguous-connection-reset-replay-boundary.
 */
export const UPSTREAM_RESET_REPLAY_REFUSED_CODE = "upstream_reset_replay_refused";
const NON_REPLAYABLE_UPSTREAM_CODES: ReadonlySet<string> = new Set([
  UPSTREAM_NO_RESPONSE_CODE,
  UPSTREAM_CLOSED_BEFORE_RESPONSE_CODE,
  UPSTREAM_RESET_REPLAY_REFUSED_CODE,
]);

export function isNonReplayableUpstreamCode(code: unknown): boolean {
  return typeof code === "string" && NON_REPLAYABLE_UPSTREAM_CODES.has(code);
}

/**
 * True for the one non-replayable code this proxy owns end to end. The status it carries is
 * a local decision, so a re-wrapping formatter must restate it rather than inherit the
 * caller's upstream-shaped status.
 */
export function isReplayRefusalCode(code: unknown): boolean {
  return code === UPSTREAM_RESET_REPLAY_REFUSED_CODE;
}

/** Client-facing status for {@link UPSTREAM_RESET_REPLAY_REFUSED_CODE}. */
export const REPLAY_REFUSED_STATUS = 429;

/**
 * The header every surface attaches to a replay refusal, and its only accepted value.
 *
 * Dropping `Retry-After` is necessary and not sufficient. The Stainless-generated clients --
 * `openai` and `anthropic` for both Python and Node, which is what most callers of this proxy
 * actually are -- decide from a status table (408, 409, 429 and every 5xx) and compute their own
 * backoff when no wait is named, so a 429 with no header is still resent. `x-should-retry` is
 * the one signal each of them reads BEFORE that table, and `"false"` is the exact string they
 * compare against.
 */
export const REPLAY_REFUSAL_NO_RETRY_HEADER = "x-should-retry";
export const REPLAY_REFUSAL_NO_RETRY_VALUE = "false";

/** Spreadable form for the surfaces that build their headers as an object literal. */
export const REPLAY_REFUSAL_CLIENT_HEADERS: Readonly<Record<string, string>> = Object.freeze({
  [REPLAY_REFUSAL_NO_RETRY_HEADER]: REPLAY_REFUSAL_NO_RETRY_VALUE,
});

/**
 * Apply the one client-facing retry policy a refusal carries: no wait, and no automatic resend.
 *
 * Kept as a single function rather than two rules each surface repeats, because the two halves
 * are only correct together -- a surface that removed the wait but not the suppression still
 * hands a retrying client a turn it may already have run.
 */
export function applyReplayRefusalClientHeaders(headers: Headers): void {
  headers.delete("retry-after");
  headers.set(REPLAY_REFUSAL_NO_RETRY_HEADER, REPLAY_REFUSAL_NO_RETRY_VALUE);
}

/**
 * Mark a response that re-wraps a refusal as the same refusal.
 *
 * The verdict has to be a property of the result the surfaces pass around, because the thing it
 * would otherwise be read from is the status, and 429 is exactly what a refusal and a real rate
 * limit have in common. Every formatter between the helper that made the refusal and the client
 * builds a new Response, so each of them restates the verdict rather than dropping it.
 */
export function retainReplayRefusal<T extends Response>(response: T): T {
  markResponseNonReplayable(response);
  markReplayRefusalResponse(response);
  return response;
}

/** Carry the verdict from a response onto the one that replaces it. */
export function carryReplayRefusal<T extends Response>(source: Response, rewrapped: T): T {
  if (sameTargetRetryHolds.has(source)) sameTargetRetryHolds.add(rewrapped);
  return isReplayRefusalResponse(source) ? retainReplayRefusal(rewrapped) : rewrapped;
}

// 1 initial + 2 retries: the pool may hold more than one stale socket.
const RESET_RETRY_MAX_ATTEMPTS = 3;
const RESET_RETRY_BASE_DELAY_MS = 150;
const RESET_RETRY_MAX_DELAY_MS = 1_000;

// Transient-5xx status retry layer (pre-stream only; devlog/_plan/260716_claudecode_hardening/010).
/** Total sends one transient-retry helper call may make: 1 initial + 2 retries. */
export const TRANSIENT_RETRY_MAX_ATTEMPTS = 3;

/**
 * Transient sends already spent by one LOGICAL request.
 *
 * A mutable holder rather than a counter local to one call frame, because the thing that has to
 * share it spans frames: a combo parent runs a separate child turn per target, and a per-child
 * counter is what let one logical request reach upstream three times per target (#4546).
 */
export interface TransientSendBudget {
  used: number;
}

export function createTransientSendBudget(): TransientSendBudget {
  return { used: 0 };
}

/**
 * Refusal raised when a logical request has no send left (#4546, REQ-B04/B05).
 *
 * It is deliberately a distinct type rather than a generic `Error`: every call site that
 * catches a helper rejection today launders it into HTTP 502 `upstream_error`, which would
 * report a proxy-side budget decision as an upstream fault and hide the real 401/429 the
 * request already had. Callers must recognise this and return the structured local error
 * instead. It is a backstop, not the policy -- a call site that still holds a reusable
 * upstream response is supposed to check the remainder BEFORE it cancels that body.
 */
export class SendBudgetExhaustedError extends Error {
  readonly code = "request_send_budget_exhausted";
  constructor(label?: ErrorDestinationLabel) {
    super(label !== undefined
      ? `request send budget exhausted before dispatch (${errorDestinationText(label)})`
      : "request send budget exhausted before dispatch");
    this.name = "SendBudgetExhaustedError";
  }
}

/**
 * Configuration refusal for an attempts value that is not a send count.
 *
 * `undefined` means "use the policy default" and `0` means "refuse". A negative, fractional,
 * NaN or infinite value is a programming or configuration error, and silently substituting the
 * default for it is how a broken budget turns back into three free sends.
 */
export class InvalidSendBudgetError extends Error {
  constructor(value: unknown) {
    super(`invalid upstream send budget: ${typeof value === "number" && Number.isFinite(value) && value >= 0 ? value : "invalid"}`);
    this.name = "InvalidSendBudgetError";
  }
}

function normalizeSendAttempts(value: number | undefined, fallback: number): number {
  if (value === undefined) return fallback;
  if (!Number.isInteger(value) || value < 0) throw new InvalidSendBudgetError(value);
  return value;
}
const TRANSIENT_RETRY_BASE_DELAY_MS = 400;
const TRANSIENT_RETRY_MAX_DELAY_MS = 5_000;
// A failed attempt slower than this is the "slow 502" incident shape (191s observed on
// 2026-07-15): retrying it only duplicates upstream load past client timeouts — return it.
const TRANSIENT_RETRY_SLOW_ATTEMPT_MS = 15_000;

/**
 * Upstream statuses treated as transient: gateway errors and Cloudflare 52x.
 * 500 is included per the OpenAI SDK default (auto-retries >=500; Tier-2 proven in
 * devlog/260716_ocx_claude_sol_502_midstream/02). 507 was observed in the 48h ledger
 * but is deliberately excluded (storage-class, not gateway-transient).
 */
export function isTransientUpstreamStatus(status: number): boolean {
  return status === 500 || status === 502 || status === 503 || status === 504
    || status === 520 || status === 521 || status === 522;
}

export interface RetryBackoffOptions {
  baseDelayMs: number;
  maxDelayMs: number;
  headers?: Headers;
  /**
   * Treat a provider's `Retry-After` as the earliest legal send rather than something the
   * local maximum may shorten. Opt-in per caller so the change lands on the transient path
   * first instead of silently lengthening every adapter's backoff.
   */
  retryAfterIsLowerBound?: boolean;
  /**
   * The wait deadline a caller applies to an honoured `Retry-After`. The delay itself is
   * never shortened: an instruction longer than the deadline is a reason to END with the
   * upstream answer, not to send early. Kept for callers that still pass it.
   */
  retryAfterCeilingMs?: number;
}

/** One minute, matching the same-target 429 ceiling the key-failover path already uses. */
export const RETRY_AFTER_CEILING_MS = 60_000;

export function abortError(signal?: AbortSignal): unknown {
  return signal?.reason ?? new DOMException("The operation was aborted", "AbortError");
}

export async function sleepWithAbort(ms: number, signal?: AbortSignal): Promise<void> {
  if (ms <= 0) return;
  if (signal?.aborted) throw abortError(signal);
  await new Promise<void>((resolve, reject) => {
    let timer: ReturnType<typeof setTimeout>;
    const cleanup = () => {
      clearTimeout(timer);
      signal?.removeEventListener("abort", onAbort);
    };
    const onAbort = () => {
      cleanup();
      reject(abortError(signal));
    };
    timer = setTimeout(() => {
      cleanup();
      resolve();
    }, ms);
    signal?.addEventListener("abort", onAbort, { once: true });
  });
}

/**
 * Best-effort, bounded cancellation of a response body before a retry backoff.
 *
 * The 429 paths release the unread body before waiting so sockets do not accumulate under a
 * rate-limit storm, but a never-settling `cancel()` promise must not be able to block the
 * abort-aware backoff (client cancel, `maxIntervalMs`, or the cumulative header deadline).
 * Cancellation is started and its rejection observed; the await is bounded by `timeoutMs`
 * and the abort signal. This mirrors the rotation-path guarantee (release is initiated, not
 * awaited forever) while preserving the resource-release intent of the same-target paths.
 */
export async function releaseResponseBodyBestEffort(
  body: ReadableStream<Uint8Array> | null,
  signal: AbortSignal | undefined,
  timeoutMs = 1_000,
): Promise<void> {
  if (!body) return;
  if (signal?.aborted) {
    void body.cancel().catch(() => {});
    return;
  }
  const cancel = body.cancel().catch(() => {});
  if (!signal) {
    let timer: ReturnType<typeof setTimeout> | undefined;
    try {
      await Promise.race([cancel, new Promise<void>(resolve => { timer = setTimeout(resolve, timeoutMs); })]);
    } finally {
      clearTimeout(timer);
    }
    return;
  }
  await new Promise<void>(resolve => {
    let timer: ReturnType<typeof setTimeout>;
    /**
     * Abort hook: clear the bounded-body release timer and settle the promise so a
     * never-settling cancel() can never block the abort-aware backoff.
     */
    const onAbort = () => {
      clearTimeout(timer);
      resolve();
    };
    timer = setTimeout(() => {
      signal.removeEventListener("abort", onAbort);
      resolve();
    }, timeoutMs);
    signal.addEventListener("abort", onAbort, { once: true });
    void cancel.then(() => {
      clearTimeout(timer);
      signal.removeEventListener("abort", onAbort);
      resolve();
    });
  });
}

/**
 * Abort-aware sleep that yields an adapter `heartbeat` at least every `heartbeatIntervalMs`.
 * The Responses bridge treats a returned iterator event as upstream liveness and aborts turns
 * that stay silent past the stall budget (default 300s), while a retryOn429 wait may legally
 * reach 600s — so deliberate waits must keep the watchdog fed or a long backoff is killed
 * mid-turn. The final chunk always yields once, which doubles as the post-wait liveness beat.
 */
export async function* sleepWithHeartbeats(
  ms: number,
  signal?: AbortSignal,
  heartbeatIntervalMs = 10_000,
): AsyncGenerator<{ type: "heartbeat" }> {
  if (ms <= 0) return;
  // Guard against a non-positive interval: a zero/negative step would spin the loop forever
  // while sleepWithAbort early-returns without ever observing the abort signal. NaN must be
  // normalized too: Math.max(1, NaN) is NaN, which would abort the wait after one beat.
  const stepMs = Number.isNaN(heartbeatIntervalMs) ? 1 : Math.max(1, heartbeatIntervalMs);
  let remaining = ms;
  while (remaining > 0) {
    const chunk = Math.min(remaining, stepMs);
    await sleepWithAbort(chunk, signal);
    remaining -= chunk;
    yield { type: "heartbeat" };
  }
}

export interface SameTarget429WaitOptions {
  body: ReadableStream<Uint8Array> | null;
  signal?: AbortSignal;
  delayMs: number;
  /**
   * When set, the wait yields adapter heartbeats so bridge stall watchdogs stay fed.
   * Omit for pre-stream recovery paths that have no stall watchdog.
   */
  heartbeatIntervalMs?: number;
}

/**
 * Shared pre-replay prep for opt-in same-target 429 waits:
 * release the unread 429 body, then sleep (optionally with heartbeats).
 * Callers still own attempt budgeting, abort re-checks, and the replay itself.
 */
export async function* prepareSameTarget429Wait(
  options: SameTarget429WaitOptions,
): AsyncGenerator<{ type: "heartbeat" }> {
  await releaseResponseBodyBestEffort(options.body, options.signal);
  if (options.heartbeatIntervalMs === undefined) {
    await sleepWithAbort(options.delayMs, options.signal);
    return;
  }
  yield* sleepWithHeartbeats(options.delayMs, options.signal, options.heartbeatIntervalMs);
}

export function isConnectionResetError(err: unknown): boolean {
  if (!(err instanceof Error)) return false;
  // Aborts and timeouts are caller decisions / honest failures — never retryable.
  if (err.name === "AbortError" || err.name === "TimeoutError") return false;
  const code = (err as { code?: unknown }).code;
  if (code === "ECONNRESET" || code === "EPIPE") return true;
  const msg = err.message.toLowerCase();
  return msg.includes("socket connection was closed unexpectedly")
    || msg.includes("connection reset by peer");
}

function retryAfterDelayMs(headers: Headers): number | undefined {
  const raw = headers.get("retry-after")?.trim();
  if (!raw) return undefined;
  const seconds = Number(raw);
  if (Number.isFinite(seconds)) return Math.max(0, seconds * 1000);
  const dateMs = Date.parse(raw);
  if (!Number.isFinite(dateMs)) return undefined;
  return Math.max(0, dateMs - Date.now());
}

export function retryBackoffDelayMs(attempt: number, opts: RetryBackoffOptions): number {
  const retryAfter = opts.headers ? retryAfterDelayMs(opts.headers) : undefined;
  const exp = Math.min(opts.baseDelayMs * (2 ** attempt), opts.maxDelayMs);
  const jittered = Math.floor(exp * (0.8 + Math.random() * 0.4));
  if (retryAfter === undefined) return jittered;
  if (opts.retryAfterIsLowerBound !== true) {
    // Historical behaviour, still the default for every caller that has not opted in.
    return Math.min(retryAfter, opts.maxDelayMs);
  }
  // A provider that names a wait is stating when it will serve again; sending earlier is a
  // request we already know will be refused, and refusing it twice is the retry storm the
  // header exists to prevent. The local maximum bounds our OWN exponential backoff and has no
  // business shortening someone else's instruction, so the instruction is returned in full.
  // Whether the request can afford to wait that long is the caller's deadline decision --
  // fetchWithTransientRetry ends with the upstream answer rather than retrying early.
  return Math.max(retryAfter, jittered);
}

export function cancelResponseBodyBestEffort(res: Response): void {
  try {
    const cancellation = res.body?.cancel();
    if (cancellation) void cancellation.catch(() => {});
  } catch {
    // Cancellation is cleanup only; retries must not wait for or fail because of it.
  }
}

/**
 * Whether an answer to a spent operator replacement would invite yet another send.
 *
 * Once the one replacement a request may spend has gone out, the first send may already have run
 * the turn, so nothing this exchange returns may cause a third send. Two parties would send again:
 * the client, whose retry table covers 408, 409, 429 and every 5xx (the Codex client retries 5xx
 * whatever the headers say; see {@link REPLAY_REFUSED_STATUS}), and this proxy, whose credential
 * and quota recovery resends on 401 (token refresh, key and pool rotation) and on 402/429
 * (account rotation). A client that follows a 307 or 308 sends the same POST body again, and a 413
 * is answered as a context overflow the client compacts and resends, so those belong here too.
 * {@link isTransientUpstreamStatus} is only the gateway subset of that set: 429 and 529 escaped
 * it. These statuses settle as the refusal instead.
 */
function invitesResendAfterReplacement(status: number): boolean {
  return status === 401 || status === 402 || status === 408 || status === 409 || status === 429
    || status === 307 || status === 308 || status === 413 || status >= 500;
}

/**
 * The answer a request keeps once its one operator replacement has gone out.
 *
 * A status that invites another send settles as the refusal. Any other answer keeps its real
 * status: no client retries it, and the caller needs the evidence (a 400 names the request
 * defect). The marker still stops this process from using it as a recovery trigger, such as the
 * opaque-blob rebuild of a 400 or a combo hop on a context overflow, because each of those checks
 * it before sending again.
 */
export function settleOperatorReplacement(response: Response): Response {
  if (response.ok) return response;
  if (invitesResendAfterReplacement(response.status)) {
    cancelResponseBodyBestEffort(response);
    return replayRefusalResponse();
  }
  markResponseNonReplayable(response);
  return response;
}

export async function fetchWithAttemptDeadline(
  url: string,
  init: RequestInit,
  timeoutMs: number,
  abortSignal?: AbortSignal,
  preferIdentityEncoding = false,
  executor: typeof globalThis.fetch = globalThis.fetch,
): Promise<Response> {
  const attemptTimeout = clearableDeadline(timeoutMs, abortSignal);
  const headers = new Headers(init.headers);
  if (preferIdentityEncoding && !headers.has("accept-encoding")) {
    headers.set("accept-encoding", "identity");
  }
  try {
    return await executor(url, {
      ...init,
      headers,
      redirect: "manual",
      signal: attemptTimeout.signal,
    });
  } finally {
    // Only the header timer is cleared. The composed signal still contains the parent, so a
    // caller abort after headers continue to cancel consumption of the returned response body.
    attemptTimeout.clear();
  }
}

export interface ResetRetryOptions {
  /**
   * Opt in only when repeating this operation cannot duplicate upstream effects.
   * This permits reset retries, not extra sends: attempts and onSendsConsumed still
   * bound and count every physical send. A string body is not replay-safety proof.
   */
  replaySafe?: boolean;
  abortSignal?: AbortSignal;
  /** Short host/path label for the retry warn log (no secrets/query strings). */
  label?: string;
  /** Client error labels are separate from the operational retry-log label. */
  destination?: ErrorDestinationLabel;
  /** Total upstream sends allowed, including the first one. Not a per-layer retry count. */
  attempts?: number;
  /**
   * Reports how many upstream sends this call actually consumed, so a caller that spans
   * several legs of one request (initial send, then a 429/account-recovery refetch) can
   * keep them on ONE budget instead of handing each leg a fresh one.
   *
   * It lives on the RESET options, not on the transient ones, because every leg that falls
   * back to reset-only retry -- the non-policy adapter initial send, and every
   * `rebuildAndRefetch` recovery kind whose provider has no transient policy -- was not merely
   * uncounted but UNCOUNTABLE: the callback existed on a type those call sites never reach.
   */
  onSendsConsumed?: ((sends: number) => void) & {
    beforeSend?: () => boolean;
    close?: () => void;
    execute?: <T>(run: () => Promise<T>) => Promise<T>;
  };
  /**
   * Spend one operator-granted replacement for a pre-header reset this helper would otherwise
   * refuse. Absent means no operator policy, which is the fail-closed answer.
   *
   * A callback rather than a count, and the difference is the whole point. A count handed to
   * each leg of a request is a count each leg holds: a rotation leg, a refresh leg and a
   * same-target 429 leg carry the same turn, so three numbers is three replacements of one
   * possibly-executed inference. The callback draws on ONE allowance held by the logical
   * request, which the post-header protocol gate draws on too.
   *
   * It never widens the send budget. A claimed replacement still has to fit inside
   * `attempts`, exactly like every other send this leg makes.
   */
  claimAmbiguousResend?: () => boolean;
}

export interface TransientRetryOptions extends ResetRetryOptions {
  /** Test seam: per-attempt slow budget override (defaults to TRANSIENT_RETRY_SLOW_ATTEMPT_MS). */
  slowAttemptMs?: number;
  /**
   * How long this caller can wait on an honoured `Retry-After`, defaulting to
   * {@link RETRY_AFTER_CEILING_MS}. It is a deadline, never a clamp: an instruction inside it
   * is slept in full, and an instruction past it ends the call with the upstream answer and
   * its `Retry-After` intact rather than sending early at a provider that already said it
   * would refuse. A caller with a shorter budget than a minute says so and is not parked past
   * it; a caller that can genuinely wait longer says so and is not cut short.
   */
  retryAfterCeilingMs?: number;
}

export type UpstreamSendRecovery = "connection-reset" | "transient-5xx";
type ReplayableFetch = (recovery?: UpstreamSendRecovery) => Promise<Response>;

/**
 * Rejection thrown by the upstream retry helpers when the terminal attempt
 * rejects after earlier attempts already produced credential-visible evidence:
 * transient 5xx responses, or a connection reset after the request was read.
 *
 * That evidence proves the host and credential path were reached, so the
 * failure must stay account-attributed even though the terminal promise looks
 * like a transport rejection (issue #914 review: mixed 5xx/reset -> rejection
 * must not be downgraded to the account-neutral pre-connection class). The
 * original rejection is preserved as `cause` so its code and message stay
 * inspectable. Extracted from PR #966 (Yuxin-Qiao) with attribution.
 */
export class UpstreamRetryEvidenceError extends Error {
  constructor(
    public readonly transientStatuses: readonly number[],
    cause: unknown,
    /** True when a connection-reset retry already reached the origin. */
    public readonly resetSeen = false,
  ) {
    const kinds: string[] = [];
    if (transientStatuses.length > 0) kinds.push("transient 5xx response(s)");
    if (resetSeen) kinds.push("a credential-visible connection reset");
    super(
      kinds.length > 0
        ? `upstream fetch failed after ${kinds.join(" and ")}`
        : "upstream fetch failed",
      { cause },
    );
    this.name = "UpstreamRetryEvidenceError";
  }
}

/**
 * Opt out of Bun's keep-alive pool after a connection-reset retry.
 *
 * Prefer the Bun fetch extension `keepalive: false` (transport-level) over
 * relying on the hop-by-hop `Connection: close` header alone — Bun has ignored
 * that header in past releases (oven-sh/bun#20492), so a header-only retry can
 * still reuse the same half-closed pooled socket. Still set Connection: close
 * as a belt-and-suspenders signal for intermediaries that honor it.
 */
export function applyUpstreamRecoveryInit<T extends RequestInit>(
  init: T,
  recovery?: UpstreamSendRecovery,
): T & { headers: Headers } {
  const headers = new Headers(init.headers);
  if (recovery !== "connection-reset") {
    return { ...init, headers };
  }
  headers.set("connection", "close");
  return { ...init, headers, keepalive: false };
}

/**
 * The refusal this proxy returns for an ambiguous reset it will not replace. The WeakSet
 * markers protect in-process recovery and the code survives JSON re-wrapping, so a combo or
 * account-recovery layer downstream cannot read it as a replayable upstream fault. The raw
 * exception is never exposed: it can carry credentials or request data.
 */
export function replayRefusalResponse(): Response {
  const response = new Response(JSON.stringify({ error: {
    type: "upstream_error",
    code: UPSTREAM_RESET_REPLAY_REFUSED_CODE,
    message: "The upstream exchange did not complete reliably. The request may already have been processed; automatic replay was stopped.",
  } }), {
    status: REPLAY_REFUSED_STATUS,
    headers: { "content-type": "application/json", ...REPLAY_REFUSAL_CLIENT_HEADERS },
  });
  return retainReplayRefusal(response);
}

/**
 * Run `doFetch` within one send budget. Connection-reset-shaped rejections are
 * terminal by default; only an explicitly replay-safe operation receives reset retries
 * with jittered backoff. HTTP responses retain the caller's existing retry policy.
 */
export async function fetchWithResetRetry(
  doFetch: ReplayableFetch,
  opts: ResetRetryOptions = {},
  firstRecovery?: UpstreamSendRecovery,
): Promise<Response> {
  const attempts = normalizeSendAttempts(opts.attempts, RESET_RETRY_MAX_ATTEMPTS);
  // Zero is zero. The old Math.max(1, ...) floor meant an exhausted budget still bought one
  // more send on every recovery leg, which is most of what made a bounded per-layer retry
  // compose into an unbounded per-request count.
  if (attempts === 0) throw new SendBudgetExhaustedError(opts.destination ?? (opts.label ? errorDestinationLabel(opts.label) : undefined));
  try {
  let lastError: unknown;
  let sawReset = false;
  // True once this leg has spent the request's operator allowance. From that point the leg
  // settles as the refusal or an unambiguous answer: a second send of a possibly-executed turn
  // is already out, and handing the client anything it would retry compounds it.
  let spentOperatorReplacement = false;
  for (let attempt = 0; attempt < attempts; attempt++) {
    if (opts.abortSignal?.aborted) throw abortError(opts.abortSignal);
    // Reported before the await, one physical send at a time: a send that rejects has still
    // been made, and this helper leaves through four exits (return, reset give-up, non-reset
    // rethrow, abort), so a per-send report is the only shape that is correct on all of them.
    if (opts.onSendsConsumed?.beforeSend && !opts.onSendsConsumed.beforeSend()) throw new SendBudgetExhaustedError(opts.destination ?? (opts.label ? errorDestinationLabel(opts.label) : undefined));
    if (!opts.onSendsConsumed?.beforeSend) opts.onSendsConsumed?.(1);
    try {
      let response: Response;
      try { const run = () => doFetch(attempt === 0 ? firstRecovery : "connection-reset");
        response = await (opts.onSendsConsumed?.execute ? opts.onSendsConsumed.execute(run) : run()); }
      finally { if (opts.onSendsConsumed?.beforeSend) opts.onSendsConsumed(1); }
      return spentOperatorReplacement ? settleOperatorReplacement(response) : response;
    } catch (err) {
      if (opts.abortSignal?.aborted) throw err;
      if (!isConnectionResetError(err)) {
        // Whatever ended the leg, an operator replacement already went out, so the first send
        // may have run the turn. Settle it as the refusal instead of throwing into a caller
        // whose transport-failure path answers with a client-retryable 502.
        if (spentOperatorReplacement) return replayRefusalResponse();
        // A reset that already reached the origin is credential-visible
        // evidence: keep it attached so the terminal rejection cannot be
        // downgraded to the pre-connection neutral class (#914 review).
        if (sawReset) throw new UpstreamRetryEvidenceError([], err, true);
        throw err;
      }
      if (opts.replaySafe === true) {
        // Repeating this operation cannot duplicate anything, so an exhausted budget rethrows
        // and the caller's own error path takes over.
        if (attempt === attempts - 1) throw err;
      } else {
        // The stage table refuses an ambiguous pre-header reset. The only thing that overrides
        // it is an operator allowance, and claiming it here is what keeps the grant single --
        // the post-header protocol gate spends the same counter for the same logical request.
        // Return evidence rather than throwing a generic transport error: outer catches
        // otherwise turn it into a replayable 502 and a combo/account recovery resends it.
        if (attempt + 1 >= attempts || opts.claimAmbiguousResend?.() !== true) {
          return replayRefusalResponse();
        }
        spentOperatorReplacement = true;
      }
      sawReset = true;
      lastError = err;
      console.warn(
        `[upstream-retry] connection reset${opts.label ? ` (${opts.label})` : ""} — ${
          spentOperatorReplacement ? "replacing" : "retrying"
        } (${attempt + 2}/${attempts})`,
      );
      await sleepWithAbort(retryBackoffDelayMs(attempt, {
        baseDelayMs: RESET_RETRY_BASE_DELAY_MS,
        maxDelayMs: RESET_RETRY_MAX_DELAY_MS,
      }), opts.abortSignal);
    }
  }
  throw lastError ?? new Error("upstream fetch failed");
  } finally { opts.onSendsConsumed?.close?.(); }
}

/** Bytes of a 5xx body read to look for a context-overflow verdict; the verdict caps its input too. */
const CONTEXT_OVERFLOW_PEEK_BYTES = 16_384;
/** A body that has not arrived by then is not evidence; the peek must never stall the retry. */
const CONTEXT_OVERFLOW_PEEK_MS = 250;
// A hypothetical, negated, or configuration-failure mention does not justify suppressing a retry.
// Inspect the unwrapped provider message, not its JSON spelling or unrelated envelope fields.
const NON_VERDICT_CONTEXT = /(?:\b(?:not|never|no|neither|nor|if|unless|whether)\b|n['\u2019]t\b)[^.!?\n]{0,100}\b(?:context(?:\s+(?:window|length))?|tokens?|(?:input|prompt)\s+is\s+too\s+long)\b|\b(?:context\s+(?:window|length)|token\s+limit)\b[^.!?\n]{0,40}\?/i;
const CONTEXT_CONFIG_FAILURE = /\b(?:fail(?:ed|ure)?|unable|cannot|error)\b[^.!?\n]{0,100}\b(?:configuration|settings|documentation)\b|\b(?:configuration|settings|documentation)\b[^.!?\n]{0,100}\b(?:fail(?:ed|ure)?|unavailable|missing|error)\b/i;

/** Up to `maxBytes` of a clone of the body, or what arrived before the deadline. */
async function peekBodyText(res: Response, maxBytes: number, deadlineMs: number, signal?: AbortSignal): Promise<string> {
  const reader = res.clone().body?.getReader();
  if (!reader) return "";
  const decoder = new TextDecoder();
  let text = "";
  let bytes = 0;
  let emptyChunks = 0;
  const deadlineAt = performance.now() + deadlineMs;
  let timer: ReturnType<typeof setTimeout> | undefined;
  let stop = () => {};
  const timedOut = new Promise<null>(resolve => {
    stop = () => resolve(null);
    timer = setTimeout(stop, deadlineMs);
    signal?.addEventListener("abort", stop, { once: true });
    if (signal?.aborted) stop();
  });
  try {
    while (bytes < maxBytes) {
      if (signal?.aborted || performance.now() >= deadlineAt) break;
      const chunk = await Promise.race([reader.read(), timedOut]);
      if (!chunk || chunk.done) break;
      if (chunk.value.byteLength === 0) {
        if (++emptyChunks % 64 === 0) await new Promise<void>(resolve => setTimeout(resolve, 0));
        continue;
      }
      emptyChunks = 0;
      const bounded = chunk.value.subarray(0, maxBytes - bytes);
      bytes += bounded.byteLength;
      text += decoder.decode(bounded, { stream: true });
    }
    return text;
  } finally {
    clearTimeout(timer);
    signal?.removeEventListener("abort", stop);
    reader.cancel().catch(() => {});
  }
}

/**
 * True when a transient-status response's body is a definite context-overflow verdict. Peeks a
 * clone within a byte and time bound, so the caller still receives the untouched body. A slow
 * or endless body keeps the normal retry unless its observed prefix already proves overflow.
 * "Too many tokens" alone is not enough here: a 5xx
 * "Too many tokens, please wait" is a throttle that the retry exists for.
 */
async function isContextOverflowResponse(res: Response, signal?: AbortSignal): Promise<boolean> {
  if ((res.headers.get("content-type") ?? "").includes("text/event-stream")) return false;
  try {
    const text = await peekBodyText(res, CONTEXT_OVERFLOW_PEEK_BYTES, CONTEXT_OVERFLOW_PEEK_MS, signal);
    return text.length > 0 && isDefiniteContextOverflow(res.status, text.replace(/too many tokens/gi, ""),
      message => !NON_VERDICT_CONTEXT.test(message) && !CONTEXT_CONFIG_FAILURE.test(message));
  } catch {
    return false;
  }
}

/**
 * fetchWithResetRetry plus the caller-selected transient-5xx policy, PRE-STREAM only.
 * A received HTTP error follows that policy; an ambiguous reset's non-replayable
 * verdict always stops it. The failed attempt's body is cancelled before the
 * retry; every returned response (ok, non-transient, aborted, slow, exhausted) keeps
 * its body intact. Honors Retry-After via retryBackoffDelayMs.
 *
 * A failed attempt slower than the slow budget is returned as-is (slow-502 shape);
 * `opts.attempts` is ONE total-send budget covering this layer and the inner reset layer
 * together, so it bounds the real number of upstream requests rather than multiplying.
 */
export async function fetchWithTransientRetry(
  doFetch: ReplayableFetch,
  opts: TransientRetryOptions = {},
): Promise<Response> {
  const budget = normalizeSendAttempts(opts.attempts, TRANSIENT_RETRY_MAX_ATTEMPTS);
  const slowAttemptMs = opts.slowAttemptMs ?? TRANSIENT_RETRY_SLOW_ATTEMPT_MS;
  const transientStatuses: number[] = [];
  // `attempts` is ONE total-send budget shared with the inner reset layer, not a per-layer
  // count. Forwarding it into every `fetchWithResetRetry` made the two multiply: with
  // `attempts: 3` the outer loop ran 3 transient rounds and each round independently retried
  // 3 connection resets, so a single call could emit 9 upstream sends — and 10 could emit 100.
  // That was harmless only because no caller passed `attempts`; the provider-level
  // `transientRetryOn5xx` policy is the first one that does, and multiplying load against an
  // already-failing provider is worse than not retrying at all.
  let sent = 0;
  const countedFetch: ReplayableFetch = (recovery) => {
    // Incremented BEFORE the await so a rejected send still consumes budget; counting only
    // successes would let a reset storm loop without bound.
    if (opts.onSendsConsumed?.beforeSend && !opts.onSendsConsumed.beforeSend()) throw new SendBudgetExhaustedError(opts.destination ?? (opts.label ? errorDestinationLabel(opts.label) : undefined));
    sent += 1;
    const run = () => doFetch(recovery);
    return opts.onSendsConsumed?.execute ? opts.onSendsConsumed.execute(run) : run();
  };
  // No floor. A spent budget hands the inner helper 0, which refuses rather than buying one
  // more send -- the loop condition alone was never enough, because every later recovery leg
  // called this helper again and the floor funded each of them.
  const remaining = () => Math.max(0, budget - sent);
  // The inner reset layer now has its own `onSendsConsumed`, and these are the same physical
  // sends `countedFetch` already counts. Forwarding the reporter down the `remaining()` path
  // would report each of them twice, which is how a four-send cap becomes a two-send cap. One
  // send is counted once, by the outermost layer that owns the budget.
  const innerResetOptions = (): ResetRetryOptions => ({
    ...opts,
    attempts: remaining(),
    onSendsConsumed: undefined,
  });
  // Reported in `finally` rather than at each exit: this function returns from five places
  // and throws from one, and a caller sharing the budget across request legs must be told the
  // real count on every one of them.
  try {
  if (budget === 0) throw new SendBudgetExhaustedError(opts.destination ?? (opts.label ? errorDestinationLabel(opts.label) : undefined));
  let attemptStart = Date.now();
  let res = await fetchWithResetRetry(countedFetch, innerResetOptions());
  for (let attempt = 0; sent < budget; attempt++) {
    // A non-replayable gateway status was settled after the request body had already left
    // for the origin; retrying it here is the automatic resend the marker exists to forbid.
    // An upstream that answers with the exact `x-should-retry: false` verdict has said the same
    // thing about its own failure, so the ladder stops and hands the response back untouched.
    if (res.ok || !isTransientUpstreamStatus(res.status) || isNonReplayableResponse(res)
      || res.headers.get(REPLAY_REFUSAL_NO_RETRY_HEADER) === REPLAY_REFUSAL_NO_RETRY_VALUE) return res;
    if (opts.abortSignal?.aborted || Date.now() - attemptStart > slowAttemptMs) return res;
    // A 5xx that says the input does not fit is deterministic: resending the same body is
    // refused the same way, so it goes back for the caller to classify (#8127).
    if (await isContextOverflowResponse(res, opts.abortSignal)) return res;
    // Checked before cancelResponseBodyBestEffort so an already-aborted caller never receives
    // a response whose body we just cancelled.
    if (opts.abortSignal?.aborted) return res;
    const instructedDelay = retryAfterDelayMs(res.headers);
    // The deadline is the CALLER'S, not this module's default. Reading the constant directly
    // broke it in both directions: a caller with a 30s budget slept the full 45s an upstream
    // asked for, and a caller that could genuinely wait 120s was handed the error back for a
    // 90s instruction it was willing to honour.
    const waitDeadlineMs = opts.retryAfterCeilingMs ?? RETRY_AFTER_CEILING_MS;
    if (instructedDelay !== undefined && instructedDelay > waitDeadlineMs) {
      // Honouring the stated wait would park this request past the deadline it can commit
      // to, and sleeping only up to the deadline is a send the provider already said it will
      // refuse. End here instead: the caller receives the upstream answer with its
      // Retry-After intact and applies its own policy, exactly as on the direct path.
      return res;
    }
    console.warn(
      `[upstream-retry] transient ${res.status}${opts.label ? ` (${opts.label})` : ""} — retrying (${sent + 1}/${budget})`,
    );
    const delay = retryBackoffDelayMs(attempt, {
      baseDelayMs: TRANSIENT_RETRY_BASE_DELAY_MS,
      maxDelayMs: TRANSIENT_RETRY_MAX_DELAY_MS,
      headers: res.headers,
      retryAfterIsLowerBound: true,
    });
    cancelResponseBodyBestEffort(res);
    // Throws on abort (see sleepWithAbort): the rejection propagates, and the body we just
    // cancelled belonged to a response we were discarding anyway.
    await sleepWithAbort(delay, opts.abortSignal);
    attemptStart = Date.now();
    transientStatuses.push(res.status);
    try {
      res = await fetchWithResetRetry(countedFetch, innerResetOptions(), "transient-5xx");
    } catch (err) {
      // Keep the prior 5xx evidence attached: the origin already responded, so
      // this rejection is not pre-connection and must not classify as neutral.
      // A budget refusal is not upstream evidence of anything and must stay recognisable.
      if (err instanceof SendBudgetExhaustedError) throw err;
      throw new UpstreamRetryEvidenceError(transientStatuses, err);
    }
  }
  // Budget exhausted: the last response is returned with its body intact.
  return res;
  } finally {
    try { opts.onSendsConsumed?.(sent); } finally { opts.onSendsConsumed?.close?.(); }
  }
}

export type ProtocolSafeRefetch = (signal?: AbortSignal) => Promise<Response>;

export interface ProtocolSafeRefetchOptions extends ResetRetryOptions {
  /** The replacement must match the response contract already selected for the client. */
  acceptResponse?: (response: Response) => boolean;
  /**
   * Spend the logical request's allowance, immediately before the replacement send.
   *
   * Asked here and not earlier so a failure this helper would refuse on its own terms -- a
   * non-reset error, a cancelled caller, a spent send budget -- cannot drain the one
   * replacement a later ambiguous reset was entitled to. False refuses the replacement.
   */
  authorize?: () => boolean;
}

/**
 * Attempt ONE caller-authorized replacement of a stream that died after the response head.
 *
 * The caller owns the proof that nothing was observed -- it comes from protocol inspection,
 * not from this module -- and owns the physical-send budget. What lives here is the part that
 * is easy to get wrong: a replacement is only usable if it is a fresh, unlocked, unread body
 * that matches the contract already promised to the client, and anything else has to be
 * cancelled and the original failure preserved.
 */
export async function refetchAfterProtocolSafeReset(
  doFetch: ProtocolSafeRefetch,
  err: unknown,
  opts: ProtocolSafeRefetchOptions = {},
): Promise<Response | null> {
  if (!isConnectionResetError(err) || opts.abortSignal?.aborted || opts.attempts === 0) return null;
  const label = opts.label
    ? " (" + redactSecretString(opts.label).replace(/[\r\n\u0000-\u001f\u007f]/g, "").slice(0, 128) + ")"
    : "";
  if (opts.authorize && !opts.authorize()) {
    console.warn("[upstream-retry] post-header reset replacement refused" + label + "; preserving original stream error");
    return null;
  }
  let replacement: Response;
  try {
    replacement = await doFetch(opts.abortSignal);
  } catch {
    console.warn("[upstream-retry] protocol-safe refetch failed" + label + "; preserving original stream error");
    return null;
  }
  const body = replacement.body;
  let accepted = !opts.abortSignal?.aborted && replacement.ok && body !== null
    && !replacement.bodyUsed && !body.locked && !isNonReplayableResponse(replacement);
  try { if (accepted && opts.acceptResponse) accepted = opts.acceptResponse(replacement); }
  catch { accepted = false; }
  if (!accepted || opts.abortSignal?.aborted || body?.locked) {
    try { void body?.cancel().catch(() => {}); } catch { /* already locked or closed */ }
    console.warn("[upstream-retry] protocol-safe refetch rejected" + label + "; preserving original stream error");
    return null;
  }
  console.warn("[upstream-retry] pre-output stream reset" + label + "; using one replacement stream");
  return replacement;
}

/**
 * Wrap a streamed body so a reset that arrives before the downstream reader has consumed a single
 * byte swaps in ONE replacement body.
 *
 * The zero-byte gate is the whole reason this wrapper exists: the caller observed nothing, which is
 * the stage where a replacement may even be considered. Every other question -- whether the operator
 * granted one, whether the request is replayable, whether the replacement is a fresh unlocked body
 * that matches the contract already promised to the client -- belongs to
 * {@link refetchAfterProtocolSafeReset}. Delegating rather than re-deciding is what keeps the chat
 * lane from drifting away from the one the Responses stream already uses.
 *
 * Partial output is never masked: once a byte has reached the caller, the original failure stands.
 */
export function wrapWithZeroOutputRefetch(
  body: ReadableStream<Uint8Array>,
  doFetch: ProtocolSafeRefetch,
  // `authorize` is optional on the shared options but required here: a zero-output replacement
  // is always a post-header resend, so every caller must name the gate that weighs it.
  opts: ProtocolSafeRefetchOptions & { authorize: () => boolean },
): ReadableStream<Uint8Array> {
  let reader = body.getReader();
  let bytesRead = 0;
  let retried = false;
  return new ReadableStream<Uint8Array>({
    async pull(controller) {
      for (;;) {
        try {
          const { done, value } = await reader.read();
          if (done) {
            controller.close();
            return;
          }
          bytesRead += value.byteLength;
          controller.enqueue(value);
          return;
        } catch (err) {
          if (!retried && bytesRead === 0 && !opts.abortSignal?.aborted) {
            retried = true;
            const replacement = await refetchAfterProtocolSafeReset(doFetch, err, { ...opts, authorize: opts.authorize });
            if (replacement?.body) {
              try { void reader.cancel().catch(() => {}); } catch { /* broken reader; the replacement won */ }
              reader = replacement.body.getReader();
              continue;
            }
          }
          try { controller.error(err); } catch { /* already torn down */ }
          return;
        }
      }
    },
    cancel(reason) {
      try { void reader.cancel(reason).catch(() => {}); } catch { /* already torn down */ }
    },
  });
}