diff --git a/src/__tests__/model-busy-retry.test.ts b/src/__tests__/model-busy-retry.test.ts
new file mode 100644
index 0000000..daf008d
--- /dev/null
+++ b/src/__tests__/model-busy-retry.test.ts
@@ -0,0 +1,570 @@
+import type {
+  Api,
+  AssistantMessage,
+  AssistantMessageEvent,
+  AssistantMessageEventStream,
+  Context,
+  Model,
+} from "@earendil-works/pi-ai";
+import { isRetryableAssistantError, retryAssistantCall } from "@earendil-works/pi-ai/utils/retry";
+import { afterEach, describe, expect, it, vi } from "vitest";
+import { parseModelBusyRejection, streamQoder } from "../protocol/stream.js";
+
+vi.mock("../auth/oauth.js", () => ({
+  resolveQoderIdentity: vi.fn().mockResolvedValue({
+    access: "fake",
+    userID: "test-user",
+    email: "test@example.com",
+    name: "Test User",
+    machineID: "test-machine",
+    refresh: "",
+    expires: 0,
+  }),
+}));
+
+function envelope(body: string, statusCodeValue = 200, statusCode = "OK"): string {
+  return (
+    "data:" +
+    JSON.stringify({
+      headers: { "Content-Type": ["application/json"] },
+      body,
+      statusCodeValue,
+      statusCode,
+    }) +
+    "\n\n"
+  );
+}
+
+const DONE_SSE = envelope("[DONE]");
+
+function chunk(delta: object, extra: object = {}): object {
+  return {
+    choices: [{ delta, index: 0 }],
+    created: 1,
+    id: "test-id",
+    model: "auto",
+    object: "chat.completion.chunk",
+    ...extra,
+  };
+}
+
+function finishChunk(finishReason: string): object {
+  return {
+    choices: [{ finish_reason: finishReason, index: 0 }],
+    created: 1,
+    id: "test-id",
+    model: "auto",
+    object: "chat.completion.chunk",
+    usage: { completion_tokens: 1, prompt_tokens: 1, total_tokens: 2 },
+  };
+}
+
+/** The exact three-layer nesting Qoder returns for a busy model. */
+function busyBody(
+  options: { retryAfterSeconds?: unknown; isQueued?: unknown; serviceAvailable?: unknown; code?: string } = {},
+): string {
+  const retryAfterSeconds = "retryAfterSeconds" in options ? options.retryAfterSeconds : 2;
+  // Absent keys are meaningful: a snapshot missing isQueued is not "ready".
+  const isQueued = "isQueued" in options ? options.isQueued : false;
+  const serviceAvailable = "serviceAvailable" in options ? options.serviceAvailable : true;
+  const code = options.code ?? "10605";
+  const queue = JSON.stringify({
+    isQueued,
+    modelKey: "ultimate",
+    queueCount: 0,
+    queueType: "p4",
+    retryAfterSeconds,
+    serviceAvailable,
+    waitTime: 0,
+  });
+  return JSON.stringify({
+    code: "403",
+    message: JSON.stringify({ code, message: queue, subCode: "RESOURCE_LIMIT" }),
+  });
+}
+
+function busySse(options: Parameters<typeof busyBody>[0] = {}): string {
+  return envelope(busyBody(options), 403, "Forbidden");
+}
+
+const SUCCESS_SSE =
+  envelope(JSON.stringify(chunk({ content: "pong", role: "assistant" }))) +
+  envelope(JSON.stringify(finishChunk("stop"))) +
+  DONE_SSE;
+
+type FetchStep = string | { body: string; status: number };
+
+function mockFetchSequence(steps: FetchStep[]): typeof fetch {
+  const queue = [...steps];
+  return vi.fn(async () => {
+    const step = queue.length > 1 ? queue.shift() : queue[0];
+    if (step === undefined) throw new Error("mock fetch queue is empty");
+    if (typeof step === "string") {
+      return new Response(step, { status: 200, headers: { "content-type": "text/event-stream" } });
+    }
+    return new Response(step.body, { status: step.status, headers: { "content-type": "text/event-stream" } });
+  }) as unknown as typeof fetch;
+}
+
+/** A chunk with no id/model so tests can prove those fields were not carried over. */
+function bareChunk(delta: object): object {
+  return { choices: [{ delta, index: 0 }], created: 1, object: "chat.completion.chunk" };
+}
+
+function makeModel(): Model<Api> {
+  return { id: "Ultimate", api: "qoder-api" as Api, provider: "qoder" } as Model<Api>;
+}
+
+function makeContext(): Context {
+  return {
+    systemPrompt: "test",
+    messages: [{ role: "user", content: "hi" }],
+    tools: [],
+  } as unknown as Context;
+}
+
+async function consume(stream: AssistantMessageEventStream): Promise<AssistantMessageEvent[]> {
+  const events: AssistantMessageEvent[] = [];
+  for await (const event of stream) {
+    events.push(event);
+    if (event.type === "done" || event.type === "error") break;
+  }
+  return events;
+}
+
+function errorOf(events: AssistantMessageEvent[]): AssistantMessage {
+  const error = events.find((event) => event.type === "error");
+  expect(error, "expected an error event").toBeDefined();
+  return (error as { error: AssistantMessage }).error;
+}
+
+function doneOf(events: AssistantMessageEvent[]): AssistantMessage {
+  const done = events.find((event) => event.type === "done");
+  expect(done, "expected a done event").toBeDefined();
+  return (done as { message: AssistantMessage }).message;
+}
+
+function decodeRequestBody(init: RequestInit | undefined): Record<string, unknown> {
+  const custom = "_doRTgHZBKcGVjlvpC,@aFSx#DPuNJme&i*MzLOEn)sUrthbf%Y^w.(kIQyXqWA!";
+  const standard = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";
+  const encoded = Buffer.from(init?.body as Uint8Array).toString("utf8");
+  const rearranged = [...encoded]
+    .map((character) => (character === "$" ? "=" : standard[custom.indexOf(character)] || character))
+    .join("");
+  const third = Math.floor(rearranged.length / 3);
+  const base64 =
+    rearranged.slice(rearranged.length - third) +
+    rearranged.slice(third, rearranged.length - third) +
+    rearranged.slice(0, third);
+  return JSON.parse(Buffer.from(base64, "base64").toString("utf8")) as Record<string, unknown>;
+}
+
+describe("parseModelBusyRejection", () => {
+  it("reads the retry hint out of the real nested envelope body", () => {
+    expect(parseModelBusyRejection(busyBody({ retryAfterSeconds: 2 }))).toEqual({ retryAfterMs: 2000 });
+  });
+
+  it("rejects a real queue instead of treating it as ready", () => {
+    expect(parseModelBusyRejection(busyBody({ isQueued: true }))).toBeUndefined();
+  });
+
+  it("rejects a snapshot without the ready flags", () => {
+    expect(parseModelBusyRejection(busyBody({ serviceAvailable: undefined }))).toBeUndefined();
+    expect(parseModelBusyRejection(busyBody({ isQueued: undefined }))).toBeUndefined();
+  });
+
+  it("requires code 10605 and treats subCode as diagnostic only", () => {
+    expect(parseModelBusyRejection(busyBody({ code: "10604" }))).toBeUndefined();
+  });
+
+  it("falls back to an immediate retry for invalid retry hints", () => {
+    for (const invalid of [-1, 0, "2", Number.NaN, null]) {
+      expect(parseModelBusyRejection(busyBody({ retryAfterSeconds: invalid }))).toEqual({ retryAfterMs: 0 });
+    }
+  });
+
+  it("survives a non-JSON body", () => {
+    expect(parseModelBusyRejection("blocked")).toBeUndefined();
+    expect(parseModelBusyRejection(undefined)).toBeUndefined();
+  });
+});
+
+describe("streamQoder model-busy recovery", () => {
+  const originalFetch = globalThis.fetch;
+  afterEach(() => {
+    globalThis.fetch = originalFetch;
+    vi.useRealTimers();
+    vi.restoreAllMocks();
+  });
+
+  it("retries a ready rejection and streams the retried reply", async () => {
+    globalThis.fetch = mockFetchSequence([busySse({ retryAfterSeconds: 0 }), SUCCESS_SSE]);
+    const events = await consume(streamQoder(makeModel(), makeContext(), { apiKey: "fake" }));
+
+    expect(vi.mocked(globalThis.fetch)).toHaveBeenCalledTimes(2);
+    expect(doneOf(events).content.find((c) => c.type === "text")).toMatchObject({ text: "pong" });
+    expect(events.filter((e) => e.type === "start")).toHaveLength(1);
+    expect(events.filter((e) => e.type === "done" || e.type === "error")).toHaveLength(1);
+  });
+
+  it("keeps request_set_id/session_id and resets request_id + chat_record_id on retry", async () => {
+    globalThis.fetch = mockFetchSequence([busySse({ retryAfterSeconds: 0 }), SUCCESS_SSE]);
+    await consume(streamQoder(makeModel(), makeContext(), { apiKey: "fake" }));
+
+    const calls = vi.mocked(globalThis.fetch).mock.calls;
+    const first = decodeRequestBody(calls[0][1]);
+    const second = decodeRequestBody(calls[1][1]);
+
+    expect(second.request_id).toBe(second.chat_record_id);
+    expect(second.request_id).not.toBe(first.request_id);
+    expect(second.chat_record_id).not.toBe(first.chat_record_id);
+    expect(second.request_set_id).toBe(first.request_set_id);
+    expect(second.session_id).toBe(first.session_id);
+    expect(second.model_config).toEqual(first.model_config);
+    expect(second.messages).toEqual(first.messages);
+    expect(second.tools).toEqual(first.tools);
+    expect(second.parameters).toEqual(first.parameters);
+  });
+
+  it("re-signs the retried body", async () => {
+    globalThis.fetch = mockFetchSequence([busySse({ retryAfterSeconds: 0 }), SUCCESS_SSE]);
+    await consume(streamQoder(makeModel(), makeContext(), { apiKey: "fake" }));
+
+    const [first, second] = vi.mocked(globalThis.fetch).mock.calls;
+    const firstInit = first[1] as RequestInit;
+    const secondInit = second[1] as RequestInit;
+    const firstHash = (firstInit.headers as Record<string, string>)["Cosy-Bodyhash"];
+    const secondHash = (secondInit.headers as Record<string, string>)["Cosy-Bodyhash"];
+    const crypto = await import("node:crypto");
+    const expected = crypto
+      .createHash("md5")
+      .update(Buffer.from(secondInit.body as Uint8Array))
+      .digest("hex");
+    expect(secondHash).toBe(expected);
+    expect(secondHash).not.toBe(firstHash);
+  });
+
+  it("exhausts the default budget with a retryable error", async () => {
+    globalThis.fetch = mockFetchSequence([busySse({ retryAfterSeconds: 0 })]);
+    const events = await consume(streamQoder(makeModel(), makeContext(), { apiKey: "fake" }));
+    const message = errorOf(events);
+
+    expect(vi.mocked(globalThis.fetch)).toHaveBeenCalledTimes(3);
+    expect(message.errorMessage).toMatch(/model is busy/);
+    expect(message.errorMessage).toMatch(/10605/);
+    expect(message.errorMessage).toMatch(/attempts=3/);
+    expect(isRetryableAssistantError(message)).toBe(true);
+  });
+
+  it("honors an explicit retry budget of 0 and 1", async () => {
+    globalThis.fetch = mockFetchSequence([busySse({ retryAfterSeconds: 0 })]);
+    let events = await consume(streamQoder(makeModel(), makeContext(), { apiKey: "fake", maxRetries: 0 }));
+    expect(vi.mocked(globalThis.fetch)).toHaveBeenCalledTimes(1);
+    expect(errorOf(events).errorMessage).toMatch(/attempts=1/);
+
+    globalThis.fetch = mockFetchSequence([busySse({ retryAfterSeconds: 0 })]);
+    events = await consume(streamQoder(makeModel(), makeContext(), { apiKey: "fake", maxRetries: 1 }));
+    expect(vi.mocked(globalThis.fetch)).toHaveBeenCalledTimes(2);
+    expect(errorOf(events).errorMessage).toMatch(/attempts=2/);
+  });
+
+  it("stops local recovery when the server asks for longer than the local cap", async () => {
+    globalThis.fetch = mockFetchSequence([busySse({ retryAfterSeconds: 30 })]);
+    const events = await consume(streamQoder(makeModel(), makeContext(), { apiKey: "fake" }));
+    const message = errorOf(events);
+
+    expect(vi.mocked(globalThis.fetch)).toHaveBeenCalledTimes(1);
+    expect(message.errorMessage).toMatch(/requested delay 30000ms/);
+    expect(message.errorMessage).toMatch(/exceeds local cap 5000ms/);
+    expect(isRetryableAssistantError(message)).toBe(true);
+  });
+
+  it("respects a smaller caller-provided maxRetryDelayMs", async () => {
+    globalThis.fetch = mockFetchSequence([busySse({ retryAfterSeconds: 3 })]);
+    const events = await consume(streamQoder(makeModel(), makeContext(), { apiKey: "fake", maxRetryDelayMs: 1000 }));
+    expect(vi.mocked(globalThis.fetch)).toHaveBeenCalledTimes(1);
+    expect(errorOf(events).errorMessage).toMatch(/exceeds local cap 1000ms/);
+  });
+
+  it("treats maxRetryDelayMs:0 as no caller cap, keeping the 5s hard cap", async () => {
+    globalThis.fetch = mockFetchSequence([busySse({ retryAfterSeconds: 0.001 }), SUCCESS_SSE]);
+    const events = await consume(streamQoder(makeModel(), makeContext(), { apiKey: "fake", maxRetryDelayMs: 0 }));
+
+    expect(vi.mocked(globalThis.fetch)).toHaveBeenCalledTimes(2);
+    expect(doneOf(events).content.find((c) => c.type === "text")).toMatchObject({ text: "pong" });
+  });
+
+  it("falls back to the hard cap for a non-finite maxRetryDelayMs", async () => {
+    globalThis.fetch = mockFetchSequence([busySse({ retryAfterSeconds: 30 })]);
+    const events = await consume(
+      streamQoder(makeModel(), makeContext(), { apiKey: "fake", maxRetryDelayMs: Number.NaN }),
+    );
+
+    expect(vi.mocked(globalThis.fetch)).toHaveBeenCalledTimes(1);
+    expect(errorOf(events).errorMessage).toMatch(/exceeds local cap 5000ms/);
+  });
+
+  it("falls back to the default retry budget for a non-finite maxRetries", async () => {
+    globalThis.fetch = mockFetchSequence([busySse({ retryAfterSeconds: 0 })]);
+    const events = await consume(streamQoder(makeModel(), makeContext(), { apiKey: "fake", maxRetries: Number.NaN }));
+
+    expect(vi.mocked(globalThis.fetch)).toHaveBeenCalledTimes(3);
+    expect(errorOf(events).errorMessage).toMatch(/attempts=3/);
+  });
+
+  it("keeps the previous behavior for a real queue", async () => {
+    globalThis.fetch = mockFetchSequence([busySse({ isQueued: true }), SUCCESS_SSE]);
+    const events = await consume(streamQoder(makeModel(), makeContext(), { apiKey: "fake" }));
+    const message = errorOf(events);
+
+    expect(vi.mocked(globalThis.fetch)).toHaveBeenCalledTimes(1);
+    expect(message.errorMessage).toMatch(/Upstream status 403/);
+  });
+
+  it("keeps the previous behavior when only subCode matches", async () => {
+    globalThis.fetch = mockFetchSequence([busySse({ code: "10604" }), SUCCESS_SSE]);
+    const events = await consume(streamQoder(makeModel(), makeContext(), { apiKey: "fake" }));
+
+    expect(vi.mocked(globalThis.fetch)).toHaveBeenCalledTimes(1);
+    expect(errorOf(events).errorMessage).toMatch(/Upstream status 403/);
+  });
+
+  it.each([
+    ["text", chunk({ content: "partial", role: "assistant" })],
+    ["thinking", chunk({ reasoning_content: "considering" })],
+    ["a buffered thinking tag", bareChunk({ content: "<thi" })],
+    ["tool arguments", chunk({ tool_calls: [{ index: 0, function: { arguments: "{" } }] })],
+    [
+      "a tool call",
+      chunk({
+        tool_calls: [{ index: 0, id: "call_1", type: "function", function: { name: "read", arguments: "{}" } }],
+      }),
+    ],
+  ])("does not retry after %s output", async (_label, delta) => {
+    const sse = envelope(JSON.stringify(delta)) + busySse({ retryAfterSeconds: 0 });
+    globalThis.fetch = mockFetchSequence([sse, SUCCESS_SSE]);
+    const events = await consume(streamQoder(makeModel(), makeContext(), { apiKey: "fake" }));
+    const message = errorOf(events);
+
+    expect(vi.mocked(globalThis.fetch)).toHaveBeenCalledTimes(1);
+    expect(message.errorMessage).toMatch(/after partial output/);
+    expect(isRetryableAssistantError(message)).toBe(false);
+    expect(events.find((e) => e.type === "done")).toBeUndefined();
+  });
+
+  it("clears metadata, finish reason and Qoder credits left by a rejected attempt", async () => {
+    const failedAttempt =
+      envelope(
+        JSON.stringify({
+          choices: [],
+          created: 1,
+          id: "metadata-only",
+          model: "ultimate",
+          object: "chat.completion.chunk",
+          usage: {
+            credits: 7,
+            original_credits: 9,
+            billable: true,
+            completion_tokens: 0,
+            prompt_tokens: 9,
+            total_tokens: 9,
+          },
+        }),
+      ) +
+      envelope(JSON.stringify(finishChunk("length"))) +
+      busySse({ retryAfterSeconds: 0 });
+    // The successful attempt carries no id/model/usage/finish_reason at all, so
+    // any value left in the message must come from the rejected attempt.
+    const bareSuccess = envelope(JSON.stringify(bareChunk({ content: "pong" }))) + DONE_SSE;
+    globalThis.fetch = mockFetchSequence([failedAttempt, bareSuccess]);
+    const events = await consume(streamQoder(makeModel(), makeContext(), { apiKey: "fake" }));
+    const message = doneOf(events);
+    const usage = message.usage as unknown as Record<string, unknown>;
+
+    expect(vi.mocked(globalThis.fetch)).toHaveBeenCalledTimes(2);
+    expect(message.stopReason).toBe("stop");
+    expect(message.responseId).toBeUndefined();
+    expect(message.responseModel).toBeUndefined();
+    expect(message.errorMessage).toBeUndefined();
+    expect(message.usage.input).toBe(0);
+    expect(message.usage.output).toBe(0);
+    expect(message.usage.totalTokens).toBe(0);
+    expect(usage.credits).toBeUndefined();
+    expect(usage.original_credits).toBeUndefined();
+    expect(usage.billable).toBeUndefined();
+    expect(usage.charged_credits).toBeUndefined();
+    expect(usage.usd_equivalent).toBeUndefined();
+  });
+
+  it("clears attempt-one metadata when the retried request fails at HTTP level", async () => {
+    const failedAttempt =
+      envelope(
+        JSON.stringify({
+          choices: [],
+          created: 1,
+          id: "metadata-only",
+          model: "ultimate",
+          object: "chat.completion.chunk",
+          usage: { credits: 7, prompt_tokens: 9, completion_tokens: 0, total_tokens: 9 },
+        }),
+      ) + busySse({ retryAfterSeconds: 0 });
+    globalThis.fetch = mockFetchSequence([failedAttempt, { body: "boom", status: 503 }]);
+    const events = await consume(streamQoder(makeModel(), makeContext(), { apiKey: "fake" }));
+    const message = errorOf(events);
+    const usage = message.usage as unknown as Record<string, unknown>;
+
+    expect(vi.mocked(globalThis.fetch)).toHaveBeenCalledTimes(2);
+    expect(message.errorMessage).toMatch(/503/);
+    expect(message.responseId).toBeUndefined();
+    expect(message.usage.totalTokens).toBe(0);
+    expect(usage.credits).toBeUndefined();
+  });
+
+  it("waits for retryAfterSeconds with fake timers", async () => {
+    vi.useFakeTimers();
+    globalThis.fetch = mockFetchSequence([busySse({ retryAfterSeconds: 2 }), SUCCESS_SSE]);
+    const pending = consume(streamQoder(makeModel(), makeContext(), { apiKey: "fake" }));
+
+    await vi.advanceTimersByTimeAsync(1999);
+    expect(vi.mocked(globalThis.fetch)).toHaveBeenCalledTimes(1);
+    await vi.advanceTimersByTimeAsync(1);
+    const events = await pending;
+
+    expect(vi.mocked(globalThis.fetch)).toHaveBeenCalledTimes(2);
+    expect(doneOf(events).content.find((c) => c.type === "text")).toMatchObject({ text: "pong" });
+  });
+
+  it("does not sleep after the final failed attempt", async () => {
+    vi.useFakeTimers();
+    globalThis.fetch = mockFetchSequence([busySse({ retryAfterSeconds: 2 })]);
+    const pending = consume(streamQoder(makeModel(), makeContext(), { apiKey: "fake", maxRetries: 1 }));
+
+    await vi.advanceTimersByTimeAsync(2000);
+    const events = await pending;
+    expect(vi.mocked(globalThis.fetch)).toHaveBeenCalledTimes(2);
+    expect(vi.getTimerCount()).toBe(0);
+    expect(errorOf(events).errorMessage).toMatch(/attempts=2/);
+  });
+
+  it("aborts while waiting for the retry", async () => {
+    vi.useFakeTimers();
+    const controller = new AbortController();
+    globalThis.fetch = mockFetchSequence([busySse({ retryAfterSeconds: 2 }), SUCCESS_SSE]);
+    const pending = consume(streamQoder(makeModel(), makeContext(), { apiKey: "fake", signal: controller.signal }));
+
+    await vi.advanceTimersByTimeAsync(100);
+    controller.abort();
+    await vi.advanceTimersByTimeAsync(5000);
+    const events = await pending;
+
+    expect(vi.mocked(globalThis.fetch)).toHaveBeenCalledTimes(1);
+    expect(events.filter((e) => e.type === "error")).toHaveLength(1);
+    expect(errorOf(events).stopReason).toBe("aborted");
+  });
+
+  it("does not start another attempt when the abort lands while the rejected stream is cancelled", async () => {
+    const controller = new AbortController();
+    const body = new TextEncoder().encode(busySse({ retryAfterSeconds: 0 }));
+    let cancelled = false;
+    const abortingStream = new ReadableStream({
+      start(streamController) {
+        streamController.enqueue(body);
+      },
+      cancel() {
+        cancelled = true;
+        controller.abort();
+      },
+    });
+    let calls = 0;
+    globalThis.fetch = vi.fn(async () => {
+      calls += 1;
+      if (calls === 1) {
+        return new Response(abortingStream, { status: 200, headers: { "content-type": "text/event-stream" } });
+      }
+      return new Response(SUCCESS_SSE, { status: 200, headers: { "content-type": "text/event-stream" } });
+    }) as unknown as typeof fetch;
+    const events = await consume(
+      streamQoder(makeModel(), makeContext(), { apiKey: "fake", signal: controller.signal }),
+    );
+
+    expect(cancelled).toBe(true);
+    expect(calls).toBe(1);
+    expect(events.filter((e) => e.type === "error")).toHaveLength(1);
+    expect(errorOf(events).stopReason).toBe("aborted");
+  });
+
+  it("cancels the rejected response before the next attempt starts", async () => {
+    let cancelled = false;
+    let cancelObservedBeforeSecondFetch = false;
+    let calls = 0;
+    const body = new TextEncoder().encode(busySse({ retryAfterSeconds: 0 }));
+    const busyStream = new ReadableStream({
+      start(streamController) {
+        streamController.enqueue(body);
+      },
+      cancel() {
+        cancelled = true;
+      },
+    });
+    globalThis.fetch = vi.fn(async () => {
+      const index = calls;
+      calls += 1;
+      if (index === 1) {
+        cancelObservedBeforeSecondFetch = cancelled;
+        expect(busyStream.locked).toBe(false);
+      }
+      if (index === 0) {
+        return new Response(busyStream, { status: 200, headers: { "content-type": "text/event-stream" } });
+      }
+      return new Response(SUCCESS_SSE, { status: 200, headers: { "content-type": "text/event-stream" } });
+    }) as unknown as typeof fetch;
+    await consume(streamQoder(makeModel(), makeContext(), { apiKey: "fake" }));
+
+    expect(calls).toBe(2);
+    expect(cancelObservedBeforeSecondFetch).toBe(true);
+  });
+});
+
+describe("harness retry integration", () => {
+  const originalFetch = globalThis.fetch;
+  afterEach(() => {
+    globalThis.fetch = originalFetch;
+    vi.restoreAllMocks();
+  });
+
+  async function runProvider(): Promise<AssistantMessage> {
+    const events = await consume(streamQoder(makeModel(), makeContext(), { apiKey: "fake" }));
+    const terminal = events.at(-1);
+    if (!terminal) throw new Error("no terminal event");
+    if (terminal.type === "done") return terminal.message;
+    if (terminal.type === "error") return terminal.error;
+    throw new Error(`unexpected terminal event: ${terminal.type}`);
+  }
+
+  it("lets retryAssistantCall take over after the local budget", async () => {
+    const fetchMock = mockFetchSequence([
+      busySse({ retryAfterSeconds: 0 }),
+      busySse({ retryAfterSeconds: 0 }),
+      busySse({ retryAfterSeconds: 0 }),
+      SUCCESS_SSE,
+    ]);
+    globalThis.fetch = fetchMock;
+
+    const result = await retryAssistantCall(runProvider, { enabled: true, maxRetries: 1, baseDelayMs: 0 }, undefined);
+
+    // 3 provider attempts (default budget) + 1 harness retry that succeeds.
+    expect(result.stopReason).toBe("stop");
+    expect(fetchMock).toHaveBeenCalledTimes(4);
+  });
+
+  it("makes no harness retry when the harness budget is disabled", async () => {
+    const fetchMock = mockFetchSequence([busySse({ retryAfterSeconds: 0 })]);
+    globalThis.fetch = fetchMock;
+
+    const result = await retryAssistantCall(runProvider, { enabled: false, maxRetries: 3, baseDelayMs: 0 }, undefined);
+
+    expect(result.stopReason).toBe("error");
+    expect(result.errorMessage).toMatch(/model is busy/);
+    expect(fetchMock).toHaveBeenCalledTimes(3);
+  });
+});
diff --git a/src/protocol/stream.ts b/src/protocol/stream.ts
index 93add09..5d331ea 100644
--- a/src/protocol/stream.ts
+++ b/src/protocol/stream.ts
@@ -24,6 +24,93 @@ import { qoderEncodeBody } from "./encoding.js";
 import { isDegenerateDsmlTurn, stripDsmlResidue, stripThinkingTags, ThinkingTagParser } from "./thinking.js";
 import { transformMessagesForQoder, transformTools } from "./transform.js";
 
+/**
+ * Qoder's scheduler reports a busy model as error code 10605. The official
+ * client treats `isQueued:false` + `serviceAvailable:true` as "ready" and
+ * re-sends the request after `retryAfterSeconds`; without that recovery one
+ * transient rejection kills the whole turn. We mirror the bounded part of it.
+ */
+const MODEL_BUSY_CODE = "10605";
+const DEFAULT_MODEL_BUSY_RETRIES = 2;
+const MAX_MODEL_BUSY_RETRIES = 10;
+const MODEL_BUSY_MAX_WAIT_MS = 5_000;
+
+/** A recovery-eligible "model busy" rejection from Qoder's queue scheduler. */
+class QoderModelBusyError extends Error {
+  constructor(
+    message: string,
+    readonly retryAfterMs: number | undefined,
+    readonly sawModelOutput = false,
+  ) {
+    super(message);
+    this.name = "QoderModelBusyError";
+  }
+}
+
+interface ModelBusyQueueSnapshot {
+  retryAfterMs?: number;
+}
+
+/** Parse `unknown | JSON string` layers until an object is found (bounded). */
+function parseJsonObject(value: unknown, depth = 0): Record<string, unknown> | undefined {
+  if (depth > 4) return undefined;
+  if (typeof value === "string") {
+    try {
+      return parseJsonObject(JSON.parse(value), depth + 1);
+    } catch {
+      return undefined;
+    }
+  }
+  return value !== null && typeof value === "object" && !Array.isArray(value)
+    ? (value as Record<string, unknown>)
+    : undefined;
+}
+
+/**
+ * Recognize Qoder's model-busy envelope and extract the retry hint.
+ *
+ * Envelope body shape (three nested JSON layers):
+ *   {code:"403",message:"{\"code\":\"10605\",\"message\":\"{\\\"isQueued\\\":false,...}\",\"subCode\":\"RESOURCE_LIMIT\"}"}
+ *
+ * Only the documented 10605 code and the observed ready combination
+ * (`isQueued:false` + `serviceAvailable:true`) are eligible. A real queue
+ * (`isQueued:true`), missing flags, or any other code keep the previous
+ * failure behavior. `subCode` is diagnostic only — it is not a retry signal.
+ */
+export function parseModelBusyRejection(body: unknown): ModelBusyQueueSnapshot | undefined {
+  const outer = parseJsonObject(body);
+  const busy = parseJsonObject(outer?.message);
+  if (busy?.code !== MODEL_BUSY_CODE) return undefined;
+  const queue = parseJsonObject(busy.message);
+  if (!queue) return undefined;
+  if (queue.isQueued !== false || queue.serviceAvailable !== true) return undefined;
+  const seconds = queue.retryAfterSeconds;
+  const retryAfterMs =
+    typeof seconds === "number" && Number.isFinite(seconds) && seconds > 0 ? Math.round(seconds * 1000) : 0;
+  return { retryAfterMs };
+}
+
+/** Sleep that rejects as soon as the caller aborts. */
+function sleepWithAbort(ms: number, signal?: AbortSignal): Promise<void> {
+  if (signal?.aborted) return Promise.reject(new Error("Aborted"));
+  if (ms <= 0) return Promise.resolve();
+  return new Promise((resolve, reject) => {
+    if (signal?.aborted) {
+      reject(new Error("Aborted"));
+      return;
+    }
+    const onAbort = () => {
+      clearTimeout(timer);
+      reject(new Error("Aborted"));
+    };
+    const timer = setTimeout(() => {
+      signal?.removeEventListener("abort", onAbort);
+      resolve();
+    }, ms);
+    signal?.addEventListener("abort", onAbort, { once: true });
+  });
+}
+
 interface ToolCallCandidates {
   states: Set<ToolCallState>;
   complete: Set<ToolCallState>;
@@ -535,12 +622,33 @@ export function streamQoder(
         },
       };
 
-      const bodyBytes = Buffer.from(JSON.stringify(reqBody));
-      const encodedBytes = qoderEncodeBody(bodyBytes);
-
       const chatURL = getQoderChatURL(providerMode);
+      const modelSource = modelConfig.source || "system";
 
-      const headers = buildAuthHeaders(encodedBytes, chatURL, {
+      // A caller-provided retry budget (including an explicit 0) wins; otherwise
+      // retry twice, capped at qodercli's own recovery limit. Non-finite values
+      // fall back to the default so the caps stay finite.
+      const requestedRetries = options?.maxRetries;
+      const modelBusyRetries = Math.min(
+        Math.max(
+          0,
+          typeof requestedRetries === "number" && Number.isFinite(requestedRetries)
+            ? Math.floor(requestedRetries)
+            : DEFAULT_MODEL_BUSY_RETRIES,
+        ),
+        MAX_MODEL_BUSY_RETRIES,
+      );
+      // pi's contract: `maxRetryDelayMs: 0` disables the caller cap. The
+      // provider's own 5s cap still applies, so a positive retryAfterSeconds is
+      // never truncated to zero.
+      const requestedMaxWaitMs = options?.maxRetryDelayMs;
+      const modelBusyMaxWaitMs =
+        typeof requestedMaxWaitMs === "number" && Number.isFinite(requestedMaxWaitMs) && requestedMaxWaitMs > 0
+          ? Math.min(MODEL_BUSY_MAX_WAIT_MS, requestedMaxWaitMs)
+          : MODEL_BUSY_MAX_WAIT_MS;
+
+      let encodedBytes = qoderEncodeBody(Buffer.from(JSON.stringify(reqBody)));
+      let headers = buildAuthHeaders(encodedBytes, chatURL, {
         userID,
         authToken: accessToken,
         name,
@@ -548,419 +656,503 @@ export function streamQoder(
         machineID,
       });
 
-      const modelSource = modelConfig.source || "system";
-
-      const response = await fetch(chatURL, {
-        method: "POST",
-        headers: {
-          "Content-Type": "application/json",
-          Accept: "text/event-stream",
-          "Cache-Control": "no-cache",
-          "Accept-Encoding": "identity",
-          "X-Model-Key": qoderModel,
-          "X-Model-Source": modelSource,
-          ...headers,
-        },
-        body: encodedBytes as unknown as BodyInit,
-        signal: options?.signal,
-      });
+      const thinkingEnabled = (options?.reasoning as unknown) !== false && (options?.reasoning as unknown) !== "off";
 
-      if (!response.ok) {
-        const errText = await response.text();
-        throw new Error(`Qoder API request failed: ${response.status} ${response.statusText}. Response: ${errText}`);
-      }
+      // Exactly one `start` per logical stream: recovery retries reuse it.
+      stream.push({ type: "start", partial: output });
 
-      const reader = response.body?.getReader();
-      if (!reader) throw new Error("No response body");
-      const decoder = new TextDecoder();
-      let buffer = "";
-      let bufferStart = 0;
+      const runAttempt = async (): Promise<void> => {
+        // Reset every mutable output field BEFORE the request: a failed attempt
+        // may otherwise leak metadata, usage (including Qoder credits) or a
+        // finish_reason into the attempt that succeeds, or into the terminal
+        // error when the next attempt fails at fetch/HTTP level.
+        options?.signal?.throwIfAborted();
+        output.stopReason = "stop";
+        output.errorMessage = undefined;
+        output.responseId = undefined;
+        output.responseModel = undefined;
+        output.usage = {
+          input: 0,
+          output: 0,
+          cacheRead: 0,
+          cacheWrite: 0,
+          totalTokens: 0,
+          cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
+        };
+
+        const response = await fetch(chatURL, {
+          method: "POST",
+          headers: {
+            "Content-Type": "application/json",
+            Accept: "text/event-stream",
+            "Cache-Control": "no-cache",
+            "Accept-Encoding": "identity",
+            "X-Model-Key": qoderModel,
+            "X-Model-Source": modelSource,
+            ...headers,
+          },
+          body: encodedBytes as unknown as BodyInit,
+          signal: options?.signal,
+        });
 
-      let contentBlockIndex = -1;
-      let thinkingBlockIndex = -1;
-      // Tail of the reasoning channel before DSML residue is stripped. Only the
-      // tail matters (the degenerate-turn check looks for trailing markup), so
-      // this stays bounded instead of growing with the reply.
-      let rawReasoningTail = "";
-      let toolCalls: ToolCallAggregation | undefined;
+        if (!response.ok) {
+          const errText = await response.text();
+          throw new Error(`Qoder API request failed: ${response.status} ${response.statusText}. Response: ${errText}`);
+        }
 
-      const thinkingEnabled = (options?.reasoning as unknown) !== false && (options?.reasoning as unknown) !== "off";
-      const thinkingParser = thinkingEnabled ? new ThinkingTagParser(output, stream) : null;
+        const reader = response.body?.getReader();
+        if (!reader) throw new Error("No response body");
+        const decoder = new TextDecoder();
+        let buffer = "";
+        let bufferStart = 0;
 
-      stream.push({ type: "start", partial: output });
+        let contentBlockIndex = -1;
+        let thinkingBlockIndex = -1;
+        // Tail of the reasoning channel before DSML residue is stripped. Only the
+        // tail matters (the degenerate-turn check looks for trailing markup), so
+        // this stays bounded instead of growing with the reply.
+        let rawReasoningTail = "";
+        let toolCalls: ToolCallAggregation | undefined;
 
-      // `data: [DONE]` is the end of the response. Break the read loop too, not
-      // just the line loop: Qoder's gateway keeps the HTTP body open after the
-      // sentinel, so waiting for `done` from reader.read() hung until the
-      // server or the OS eventually closed the socket. The full reply had
-      // already been streamed by then, so the agent looked stuck with no error.
-      let sawDone = false;
+        const thinkingParser = thinkingEnabled ? new ThinkingTagParser(output, stream) : null;
 
-      try {
-        while (!sawDone) {
-          const { done, value } = await reader.read();
-          if (done) break;
-
-          // Drop consumed prefix before appending so we do not keep growing a
-          // dead head of the string across chunks.
-          if (bufferStart > 0) {
-            buffer = buffer.substring(bufferStart);
-            bufferStart = 0;
-          }
-          buffer += decoder.decode(value, { stream: true });
+        let sawModelOutput = false;
 
-          while (true) {
-            const lineEnd = buffer.indexOf("\n", bufferStart);
-            if (lineEnd === -1) break;
+        // `data: [DONE]` is the end of the response. Break the read loop too, not
+        // just the line loop: Qoder's gateway keeps the HTTP body open after the
+        // sentinel, so waiting for `done` from reader.read() hung until the
+        // server or the OS eventually closed the socket. The full reply had
+        // already been streamed by then, so the agent looked stuck with no error.
+        let sawDone = false;
 
-            const line = buffer.substring(bufferStart, lineEnd).trim();
-            bufferStart = lineEnd + 1;
+        try {
+          while (!sawDone) {
+            const { done, value } = await reader.read();
+            if (done) break;
+
+            // Drop consumed prefix before appending so we do not keep growing a
+            // dead head of the string across chunks.
+            if (bufferStart > 0) {
+              buffer = buffer.substring(bufferStart);
+              bufferStart = 0;
+            }
+            buffer += decoder.decode(value, { stream: true });
 
-            if (!line.startsWith("data:")) continue;
+            while (true) {
+              const lineEnd = buffer.indexOf("\n", bufferStart);
+              if (lineEnd === -1) break;
 
-            const dataStr = line.substring(5).trim();
-            if (dataStr === "[DONE]") {
-              sawDone = true;
-              break;
-            }
+              const line = buffer.substring(bufferStart, lineEnd).trim();
+              bufferStart = lineEnd + 1;
 
-            try {
-              const envelope = JSON.parse(dataStr);
-              if (envelope.statusCodeValue && envelope.statusCodeValue !== 200) {
-                throw new Error(`Upstream status ${envelope.statusCodeValue}: ${envelope.body}`);
-              }
+              if (!line.startsWith("data:")) continue;
 
-              const innerStr = envelope.body;
-              // The gateway sends the sentinel wrapped in an envelope
-              // (`body: "[DONE]"`) as well as bare, and both mean the reply is
-              // over, so both have to end the read loop.
-              if (innerStr === "[DONE]") {
+              const dataStr = line.substring(5).trim();
+              if (dataStr === "[DONE]") {
                 sawDone = true;
                 break;
               }
-              if (!innerStr) continue;
-
-              const inner = JSON.parse(innerStr);
-              if (inner.id) output.responseId = inner.id as string;
-              if (inner.model) output.responseModel = inner.model as string;
-              if (inner.usage) {
-                applyQoderCredits(output.usage, inner.usage, model.provider === "qoder-cn" ? "qoder-cn" : "qoder");
-                const u = inner.usage as {
-                  prompt_tokens?: number;
-                  completion_tokens?: number;
-                  total_tokens?: number;
-                  completion_tokens_details?: { reasoning_tokens?: number };
-                  prompt_tokens_details?: {
-                    cacheable_tokens?: number;
-                    cached_tokens?: number;
-                    cache_write_tokens?: number;
-                  };
-                };
-                // pi-core computes `promptTokens = input + cacheRead + cacheWrite`
-                // (Anthropic convention: `input` EXCLUDES cached/written tokens).
-                // Qoder follows OpenAI semantics where `prompt_tokens` INCLUDES
-                // `cached_tokens`, so subtract cacheRead (and cache_write_tokens
-                // when reported) to match the contract pi-ai's own OpenAI
-                // provider uses. `cacheable_tokens` is a capacity metric, not a
-                // write count (it is 0 even on first-turn writes), so it is NOT
-                // mapped to cacheWrite.
-                const promptTokens = u.prompt_tokens ?? 0;
-                const cacheReadTokens = u.prompt_tokens_details?.cached_tokens ?? 0;
-                const cacheWriteTokens = u.prompt_tokens_details?.cache_write_tokens ?? 0;
-                output.usage.input = Math.max(0, promptTokens - cacheReadTokens - cacheWriteTokens);
-                output.usage.output = u.completion_tokens ?? 0;
-                output.usage.totalTokens = u.total_tokens ?? 0;
-                output.usage.cacheRead = cacheReadTokens;
-                output.usage.cacheWrite = cacheWriteTokens;
-              }
-              if (inner.choices && inner.choices.length > 0) {
-                const choice = inner.choices[0];
-                const delta = choice.delta;
-
-                if (delta) {
-                  // 1. Process reasoning/thinking content (API reasoning)
-                  if (delta.reasoning_content) {
-                    // Qoder's backend sometimes routes a literal `<thinking>`
-                    // opener into reasoning_content (with the matching
-                    // `</thinking>` closer landing in the content stream), and
-                    // sometimes leaks unparsed DSML tool-call markup there. Strip
-                    // both so the thinking block stays clean, matching the SDK's
-                    // ContentBlock model.
-                    rawReasoningTail = (rawReasoningTail + delta.reasoning_content).slice(-512);
-                    const reasoningChunk = stripDsmlResidue(stripThinkingTags(delta.reasoning_content));
-                    if (reasoningChunk) {
-                      if (thinkingBlockIndex === -1) {
-                        thinkingBlockIndex = output.content.length;
-                        output.content.push({ type: "thinking", thinking: "" });
-                        stream.push({ type: "thinking_start", contentIndex: thinkingBlockIndex, partial: output });
-                      }
-                      const block = output.content[thinkingBlockIndex] as ThinkingContent;
-                      block.thinking += reasoningChunk;
-                      stream.push({
-                        type: "thinking_delta",
-                        contentIndex: thinkingBlockIndex,
-                        delta: reasoningChunk,
-                        partial: output,
-                      });
-                    }
+
+              try {
+                const envelope = JSON.parse(dataStr);
+                if (envelope.statusCodeValue && envelope.statusCodeValue !== 200) {
+                  const busy = parseModelBusyRejection(envelope.body);
+                  if (busy) {
+                    throw new QoderModelBusyError(
+                      `Upstream status ${envelope.statusCodeValue}: ${envelope.body}`,
+                      busy.retryAfterMs,
+                      sawModelOutput,
+                    );
                   }
+                  throw new Error(`Upstream status ${envelope.statusCodeValue}: ${envelope.body}`);
+                }
 
-                  // 2. Process text content
-                  if (delta.content) {
-                    // End API thinking block if active
-                    if (thinkingBlockIndex !== -1) {
-                      const block = output.content[thinkingBlockIndex] as ThinkingContent;
-                      stream.push({
-                        type: "thinking_end",
-                        contentIndex: thinkingBlockIndex,
-                        content: block.thinking,
-                        partial: output,
-                      });
-                      thinkingBlockIndex = -1;
+                const innerStr = envelope.body;
+                // The gateway sends the sentinel wrapped in an envelope
+                // (`body: "[DONE]"`) as well as bare, and both mean the reply is
+                // over, so both have to end the read loop.
+                if (innerStr === "[DONE]") {
+                  sawDone = true;
+                  break;
+                }
+                if (!innerStr) continue;
+
+                const inner = JSON.parse(innerStr);
+                if (inner.id) output.responseId = inner.id as string;
+                if (inner.model) output.responseModel = inner.model as string;
+                if (inner.usage) {
+                  applyQoderCredits(output.usage, inner.usage, model.provider === "qoder-cn" ? "qoder-cn" : "qoder");
+                  const u = inner.usage as {
+                    prompt_tokens?: number;
+                    completion_tokens?: number;
+                    total_tokens?: number;
+                    completion_tokens_details?: { reasoning_tokens?: number };
+                    prompt_tokens_details?: {
+                      cacheable_tokens?: number;
+                      cached_tokens?: number;
+                      cache_write_tokens?: number;
+                    };
+                  };
+                  // pi-core computes `promptTokens = input + cacheRead + cacheWrite`
+                  // (Anthropic convention: `input` EXCLUDES cached/written tokens).
+                  // Qoder follows OpenAI semantics where `prompt_tokens` INCLUDES
+                  // `cached_tokens`, so subtract cacheRead (and cache_write_tokens
+                  // when reported) to match the contract pi-ai's own OpenAI
+                  // provider uses. `cacheable_tokens` is a capacity metric, not a
+                  // write count (it is 0 even on first-turn writes), so it is NOT
+                  // mapped to cacheWrite.
+                  const promptTokens = u.prompt_tokens ?? 0;
+                  const cacheReadTokens = u.prompt_tokens_details?.cached_tokens ?? 0;
+                  const cacheWriteTokens = u.prompt_tokens_details?.cache_write_tokens ?? 0;
+                  output.usage.input = Math.max(0, promptTokens - cacheReadTokens - cacheWriteTokens);
+                  output.usage.output = u.completion_tokens ?? 0;
+                  output.usage.totalTokens = u.total_tokens ?? 0;
+                  output.usage.cacheRead = cacheReadTokens;
+                  output.usage.cacheWrite = cacheWriteTokens;
+                }
+                if (inner.choices && inner.choices.length > 0) {
+                  const choice = inner.choices[0];
+                  const delta = choice.delta;
+
+                  if (delta) {
+                    // Any raw generation field ends recovery for this stream: a
+                    // retry could duplicate text, thinking, or tool arguments
+                    // (including fragments still buffered in the tag parser).
+                    if (
+                      delta.reasoning_content ||
+                      delta.content ||
+                      (Array.isArray(delta.tool_calls) && delta.tool_calls.length > 0)
+                    ) {
+                      sawModelOutput = true;
                     }
-
-                    if (thinkingParser) {
-                      thinkingParser.processChunk(delta.content);
-                      if (toolCalls?.lastToolState) refreshToolIndexes(output, toolCalls);
-                    } else {
-                      if (contentBlockIndex === -1) {
-                        contentBlockIndex = output.content.length;
-                        output.content.push({ type: "text", text: "" });
-                        stream.push({ type: "text_start", contentIndex: contentBlockIndex, partial: output });
+                    // 1. Process reasoning/thinking content (API reasoning)
+                    if (delta.reasoning_content) {
+                      // Qoder's backend sometimes routes a literal `<thinking>`
+                      // opener into reasoning_content (with the matching
+                      // `</thinking>` closer landing in the content stream), and
+                      // sometimes leaks unparsed DSML tool-call markup there. Strip
+                      // both so the thinking block stays clean, matching the SDK's
+                      // ContentBlock model.
+                      rawReasoningTail = (rawReasoningTail + delta.reasoning_content).slice(-512);
+                      const reasoningChunk = stripDsmlResidue(stripThinkingTags(delta.reasoning_content));
+                      if (reasoningChunk) {
+                        if (thinkingBlockIndex === -1) {
+                          thinkingBlockIndex = output.content.length;
+                          output.content.push({ type: "thinking", thinking: "" });
+                          stream.push({ type: "thinking_start", contentIndex: thinkingBlockIndex, partial: output });
+                        }
+                        const block = output.content[thinkingBlockIndex] as ThinkingContent;
+                        block.thinking += reasoningChunk;
+                        stream.push({
+                          type: "thinking_delta",
+                          contentIndex: thinkingBlockIndex,
+                          delta: reasoningChunk,
+                          partial: output,
+                        });
                       }
-                      const block = output.content[contentBlockIndex] as TextContent;
-                      block.text += delta.content;
-                      stream.push({
-                        type: "text_delta",
-                        contentIndex: contentBlockIndex,
-                        delta: delta.content,
-                        partial: output,
-                      });
                     }
-                  }
 
-                  // 3. Process tool calls
-                  if (delta.tool_calls && Array.isArray(delta.tool_calls)) {
-                    for (const tc of delta.tool_calls) {
-                      const id = tc.id || "";
-                      const name = tc.function?.name || "";
-                      const argDelta = tc.function?.arguments ?? "";
-                      if (typeof id !== "string" || typeof name !== "string") {
-                        throw new Error("Tool-call identity conflict: invalid metadata");
+                    // 2. Process text content
+                    if (delta.content) {
+                      // End API thinking block if active
+                      if (thinkingBlockIndex !== -1) {
+                        const block = output.content[thinkingBlockIndex] as ThinkingContent;
+                        stream.push({
+                          type: "thinking_end",
+                          contentIndex: thinkingBlockIndex,
+                          content: block.thinking,
+                          partial: output,
+                        });
+                        thinkingBlockIndex = -1;
                       }
-                      if (typeof argDelta !== "string") throw new Error("Invalid tool-call argument fragment");
-                      if (!id && !name && !argDelta) continue;
-
-                      toolCalls ??= {
-                        states: [],
-                        byId: new Map(),
-                        byIndex: new Map(),
-                        unscopedCandidates: new Set(),
-                        statesByBlock: new Map(),
-                      };
-                      const index = typeof tc.index === "number" ? tc.index : undefined;
-                      let candidates: ToolCallCandidates | undefined;
-                      if (index !== undefined) {
-                        candidates = toolCalls.byIndex.get(index);
-                        if (!candidates) {
-                          candidates = { states: new Set(), complete: new Set(), unique: undefined };
-                          toolCalls.byIndex.set(index, candidates);
+
+                      if (thinkingParser) {
+                        thinkingParser.processChunk(delta.content);
+                        if (toolCalls?.lastToolState) refreshToolIndexes(output, toolCalls);
+                      } else {
+                        if (contentBlockIndex === -1) {
+                          contentBlockIndex = output.content.length;
+                          output.content.push({ type: "text", text: "" });
+                          stream.push({ type: "text_start", contentIndex: contentBlockIndex, partial: output });
                         }
+                        const block = output.content[contentBlockIndex] as TextContent;
+                        block.text += delta.content;
+                        stream.push({
+                          type: "text_delta",
+                          contentIndex: contentBlockIndex,
+                          delta: delta.content,
+                          partial: output,
+                        });
                       }
-                      let state = id ? toolCalls.byId.get(id) : undefined;
-                      let associated = !!state && !!candidates && candidates.states.has(state);
-                      if (id && candidates && !associated && candidates.complete.size > 0) {
-                        retireCompleteCandidates(toolCalls, candidates);
-                      }
-                      const available = candidates?.states ?? toolCalls.unscopedCandidates;
-                      if (!state) {
-                        const only =
-                          available.size === 1
-                            ? candidates
-                              ? candidates.unique
-                              : available.values().next().value
-                            : undefined;
-                        if (!id && available.size > 1) {
-                          throw new Error(`Ambiguous tool-call continuation (candidates=${available.size})`);
+                    }
+
+                    // 3. Process tool calls
+                    if (delta.tool_calls && Array.isArray(delta.tool_calls)) {
+                      for (const tc of delta.tool_calls) {
+                        const id = tc.id || "";
+                        const name = tc.function?.name || "";
+                        const argDelta = tc.function?.arguments ?? "";
+                        if (typeof id !== "string" || typeof name !== "string") {
+                          throw new Error("Tool-call identity conflict: invalid metadata");
                         }
-                        // An explicit new ID never overwrites another identity.
-                        // Only a unique, still-unidentified provisional can bind.
-                        if (only && (!id || !only.id)) {
-                          state = only;
-                          associated = candidates !== undefined;
-                        } else {
-                          state = {
-                            argumentChunks: [],
-                            argumentLength: 0,
-                            depth: 0,
-                            inString: false,
-                            escaped: false,
-                            complete: false,
-                            id: "",
-                            name: "",
-                            contentIndex: -1,
-                            candidates: new Set(),
-                            unindexed: index === undefined,
-                          };
-                          toolCalls.states.push(state);
+                        if (typeof argDelta !== "string") throw new Error("Invalid tool-call argument fragment");
+                        if (!id && !name && !argDelta) continue;
+
+                        toolCalls ??= {
+                          states: [],
+                          byId: new Map(),
+                          byIndex: new Map(),
+                          unscopedCandidates: new Set(),
+                          statesByBlock: new Map(),
+                        };
+                        const index = typeof tc.index === "number" ? tc.index : undefined;
+                        let candidates: ToolCallCandidates | undefined;
+                        if (index !== undefined) {
+                          candidates = toolCalls.byIndex.get(index);
+                          if (!candidates) {
+                            candidates = { states: new Set(), complete: new Set(), unique: undefined };
+                            toolCalls.byIndex.set(index, candidates);
+                          }
                         }
-                      }
-                      if (name && state.name && name !== state.name) {
-                        throw new Error("Tool-call identity conflict: inconsistent name");
-                      }
-                      const newlyIdentified = !!id && !state.id;
-                      const newlyNamed = !!name && !state.name;
-                      if (newlyIdentified) {
-                        state.id = id;
-                        toolCalls.byId.set(id, state);
-                      }
-                      if (newlyNamed) state.name = name;
-                      if (!associated) {
-                        // Indexed continuations already have all three links.
-                        // Unindexed explicit IDs can reactivate retired states.
-                        toolCalls.unscopedCandidates.add(state);
-                        if (candidates) {
-                          state.unindexed = false;
-                          state.candidates.add(candidates);
-                          candidates.states.add(state);
-                          candidates.unique = candidates.states.size === 1 ? state : undefined;
-                          if (state.complete && state.id) candidates.complete.add(state);
+                        let state = id ? toolCalls.byId.get(id) : undefined;
+                        let associated = !!state && !!candidates && candidates.states.has(state);
+                        if (id && candidates && !associated && candidates.complete.size > 0) {
+                          retireCompleteCandidates(toolCalls, candidates);
+                        }
+                        const available = candidates?.states ?? toolCalls.unscopedCandidates;
+                        if (!state) {
+                          const only =
+                            available.size === 1
+                              ? candidates
+                                ? candidates.unique
+                                : available.values().next().value
+                              : undefined;
+                          if (!id && available.size > 1) {
+                            throw new Error(`Ambiguous tool-call continuation (candidates=${available.size})`);
+                          }
+                          // An explicit new ID never overwrites another identity.
+                          // Only a unique, still-unidentified provisional can bind.
+                          if (only && (!id || !only.id)) {
+                            state = only;
+                            associated = candidates !== undefined;
+                          } else {
+                            state = {
+                              argumentChunks: [],
+                              argumentLength: 0,
+                              depth: 0,
+                              inString: false,
+                              escaped: false,
+                              complete: false,
+                              id: "",
+                              name: "",
+                              contentIndex: -1,
+                              candidates: new Set(),
+                              unindexed: index === undefined,
+                            };
+                            toolCalls.states.push(state);
+                          }
+                        }
+                        if (name && state.name && name !== state.name) {
+                          throw new Error("Tool-call identity conflict: inconsistent name");
+                        }
+                        const newlyIdentified = !!id && !state.id;
+                        const newlyNamed = !!name && !state.name;
+                        if (newlyIdentified) {
+                          state.id = id;
+                          toolCalls.byId.set(id, state);
+                        }
+                        if (newlyNamed) state.name = name;
+                        if (!associated) {
+                          // Indexed continuations already have all three links.
+                          // Unindexed explicit IDs can reactivate retired states.
+                          toolCalls.unscopedCandidates.add(state);
+                          if (candidates) {
+                            state.unindexed = false;
+                            state.candidates.add(candidates);
+                            candidates.states.add(state);
+                            candidates.unique = candidates.states.size === 1 ? state : undefined;
+                            if (state.complete && state.id) candidates.complete.add(state);
+                          }
                         }
-                      }
-
-                      const wasComplete = state.complete;
-                      if (argDelta) appendToolArguments(state, argDelta);
-                      if (state.complete && state.id && (!wasComplete || newlyIdentified)) {
-                        // Each association is registered at most once on closure,
-                        // not revisited on every fragment or repeated ID.
-                        for (const scope of state.candidates) scope.complete.add(state);
-                      }
 
-                      if (!state.block && (state.id || state.name)) {
-                        state.block = { type: "toolCall", id: state.id, name: state.name, arguments: {} };
-                        state.contentIndex = output.content.length;
-                        output.content.push(state.block);
-                        toolCalls.statesByBlock.set(state.block, state);
-                        toolCalls.lastToolState = state;
-                        stream.push({ type: "toolcall_start", contentIndex: state.contentIndex, partial: output });
-                        // Argument-only provisional fragments have no content
-                        // position until this block exists. Replay them in order.
-                        for (const fragment of state.argumentChunks) {
-                          stream.push({
-                            type: "toolcall_delta",
-                            contentIndex: state.contentIndex,
-                            delta: fragment,
-                            partial: output,
-                          });
+                        const wasComplete = state.complete;
+                        if (argDelta) appendToolArguments(state, argDelta);
+                        if (state.complete && state.id && (!wasComplete || newlyIdentified)) {
+                          // Each association is registered at most once on closure,
+                          // not revisited on every fragment or repeated ID.
+                          for (const scope of state.candidates) scope.complete.add(state);
                         }
-                      } else if (state.block) {
-                        if (newlyIdentified) state.block.id = state.id;
-                        if (newlyNamed) state.block.name = state.name;
-                        if (argDelta) {
-                          stream.push({
-                            type: "toolcall_delta",
-                            contentIndex: state.contentIndex,
-                            delta: argDelta,
-                            partial: output,
-                          });
+
+                        if (!state.block && (state.id || state.name)) {
+                          state.block = { type: "toolCall", id: state.id, name: state.name, arguments: {} };
+                          state.contentIndex = output.content.length;
+                          output.content.push(state.block);
+                          toolCalls.statesByBlock.set(state.block, state);
+                          toolCalls.lastToolState = state;
+                          stream.push({ type: "toolcall_start", contentIndex: state.contentIndex, partial: output });
+                          // Argument-only provisional fragments have no content
+                          // position until this block exists. Replay them in order.
+                          for (const fragment of state.argumentChunks) {
+                            stream.push({
+                              type: "toolcall_delta",
+                              contentIndex: state.contentIndex,
+                              delta: fragment,
+                              partial: output,
+                            });
+                          }
+                        } else if (state.block) {
+                          if (newlyIdentified) state.block.id = state.id;
+                          if (newlyNamed) state.block.name = state.name;
+                          if (argDelta) {
+                            stream.push({
+                              type: "toolcall_delta",
+                              contentIndex: state.contentIndex,
+                              delta: argDelta,
+                              partial: output,
+                            });
+                          }
                         }
                       }
                     }
                   }
-                }
 
-                if (choice.finish_reason) {
-                  // Preserve the real upstream finish_reason (e.g. "length",
-                  // "content_filter") instead of forcing "stop" later.
-                  output.stopReason = choice.finish_reason as AssistantMessage["stopReason"];
+                  if (choice.finish_reason) {
+                    // Preserve the real upstream finish_reason (e.g. "length",
+                    // "content_filter") instead of forcing "stop" later.
+                    output.stopReason = choice.finish_reason as AssistantMessage["stopReason"];
+                  }
                 }
-              }
-            } catch (e) {
-              // A single malformed SSE line shouldn't kill the stream — skip it.
-              // But a genuine upstream error (thrown below) must propagate to the
-              // outer catch and surface as stopReason="error", not be swallowed.
-              if (e instanceof SyntaxError) {
-                if (process.env.QODER_DEBUG) {
-                  console.error("[pi-provider-qoder] skipping malformed SSE line:", dataStr.slice(0, 200));
+              } catch (e) {
+                // A single malformed SSE line shouldn't kill the stream — skip it.
+                // But a genuine upstream error (thrown below) must propagate to the
+                // outer catch and surface as stopReason="error", not be swallowed.
+                if (e instanceof SyntaxError) {
+                  if (process.env.QODER_DEBUG) {
+                    console.error("[pi-provider-qoder] skipping malformed SSE line:", dataStr.slice(0, 200));
+                  }
+                  continue;
                 }
-                continue;
+                throw e;
               }
-              throw e;
             }
           }
-        }
-      } finally {
-        // Release on DONE, abort, and identity/argument failures alike.
-        try {
-          await reader.cancel().catch(() => {});
         } finally {
-          reader.releaseLock();
+          // Release on DONE, abort, and identity/argument failures alike.
+          try {
+            await reader.cancel().catch(() => {});
+          } finally {
+            reader.releaseLock();
+          }
         }
-      }
 
-      if (thinkingParser) {
-        thinkingParser.finalize();
-        if (toolCalls?.lastToolState) refreshToolIndexes(output, toolCalls);
-      }
-
-      if (thinkingBlockIndex !== -1) {
-        const block = output.content[thinkingBlockIndex] as ThinkingContent;
-        stream.push({
-          type: "thinking_end",
-          contentIndex: thinkingBlockIndex,
-          content: block.thinking,
-          partial: output,
-        });
-      }
-
-      if (toolCalls) {
-        // Validate the whole batch before publishing any successful end event.
-        for (const state of toolCalls.states) {
-          if (!state.id || !state.name) {
-            throw new Error(`Unresolved tool-call identity (characters=${state.argumentLength})`);
-          }
-          parseToolArguments(state);
+        if (thinkingParser) {
+          thinkingParser.finalize();
+          if (toolCalls?.lastToolState) refreshToolIndexes(output, toolCalls);
         }
-        // Map insertion order is the order calls first became identifiable,
-        // independent of sparse/reused upstream indexes and provisional buffers.
-        for (const [block, state] of toolCalls.statesByBlock) {
-          block.arguments = parseToolArguments(state);
+
+        if (thinkingBlockIndex !== -1) {
+          const block = output.content[thinkingBlockIndex] as ThinkingContent;
           stream.push({
-            type: "toolcall_end",
-            contentIndex: state.contentIndex,
-            toolCall: block,
+            type: "thinking_end",
+            contentIndex: thinkingBlockIndex,
+            content: block.thinking,
             partial: output,
           });
         }
 
-        if (toolCalls.statesByBlock.size > 0 && output.stopReason !== "length") {
-          output.stopReason = "toolUse";
+        if (toolCalls) {
+          // Validate the whole batch before publishing any successful end event.
+          for (const state of toolCalls.states) {
+            if (!state.id || !state.name) {
+              throw new Error(`Unresolved tool-call identity (characters=${state.argumentLength})`);
+            }
+            parseToolArguments(state);
+          }
+          // Map insertion order is the order calls first became identifiable,
+          // independent of sparse/reused upstream indexes and provisional buffers.
+          for (const [block, state] of toolCalls.statesByBlock) {
+            block.arguments = parseToolArguments(state);
+            stream.push({
+              type: "toolcall_end",
+              contentIndex: state.contentIndex,
+              toolCall: block,
+              partial: output,
+            });
+          }
+
+          if (toolCalls.statesByBlock.size > 0 && output.stopReason !== "length") {
+            output.stopReason = "toolUse";
+          }
         }
-      }
-      // A turn the gateway stripped the tool call out of has no text and no tool
-      // call, so `stop` reads as a finished task: the agent loop ends the turn
-      // and the work dies with no error to retry. Report it instead.
-      if (isDegenerateDsmlTurn(output, rawReasoningTail)) {
-        output.stopReason = "error";
-        // "server error" is deliberate: pi-ai classifies an assistant error by
-        // matching its message (RETRYABLE_PROVIDER_ERROR_PATTERN), and that
-        // phrase is what makes the turn retryable instead of a dead end.
-        output.errorMessage =
-          "Qoder server error: degenerate model output (unparseable DSML tool-call markup, no tool call executed)";
-        stream.push({ type: "error", reason: "error", error: output });
+        // A turn the gateway stripped the tool call out of has no text and no tool
+        // call, so `stop` reads as a finished task: the agent loop ends the turn
+        // and the work dies with no error to retry. Report it instead.
+        if (isDegenerateDsmlTurn(output, rawReasoningTail)) {
+          output.stopReason = "error";
+          // "server error" is deliberate: pi-ai classifies an assistant error by
+          // matching its message (RETRYABLE_PROVIDER_ERROR_PATTERN), and that
+          // phrase is what makes the turn retryable instead of a dead end.
+          output.errorMessage =
+            "Qoder server error: degenerate model output (unparseable DSML tool-call markup, no tool call executed)";
+          stream.push({ type: "error", reason: "error", error: output });
+          stream.end();
+          return;
+        }
+
+        // Otherwise keep whatever finish_reason set upstream (defaults to "stop").
+        // Never overwrite a meaningful finish_reason ("length", "content_filter",
+        // ...) with "stop".
+        stream.push({
+          type: "done",
+          reason: output.stopReason as Extract<AssistantMessage["stopReason"], "stop" | "length" | "toolUse">,
+          message: output,
+        });
         stream.end();
-        return;
-      }
+      };
 
-      // Otherwise keep whatever finish_reason set upstream (defaults to "stop").
-      // Never overwrite a meaningful finish_reason ("length", "content_filter",
-      // ...) with "stop".
-      stream.push({
-        type: "done",
-        reason: output.stopReason as Extract<AssistantMessage["stopReason"], "stop" | "length" | "toolUse">,
-        message: output,
-      });
-      stream.end();
+      for (let attempt = 0; ; attempt++) {
+        try {
+          await runAttempt();
+          return;
+        } catch (e) {
+          if (!(e instanceof QoderModelBusyError)) throw e;
+          if (e.sawModelOutput || output.content.length > 0) {
+            // Keep this wording out of pi-ai's retryable pattern: retrying a
+            // turn that already emitted content would duplicate it.
+            throw new Error("Qoder model rejected the request after partial output (10605 RESOURCE_LIMIT)");
+          }
+          const requestedWaitMs = e.retryAfterMs ?? 0;
+          if (attempt >= modelBusyRetries) {
+            throw new QoderModelBusyError(
+              `Qoder server error: model is busy (10605 RESOURCE_LIMIT, retryAfterSeconds=${Math.round(requestedWaitMs / 1000)}, attempts=${attempt + 1})`,
+              requestedWaitMs,
+            );
+          }
+          if (requestedWaitMs > modelBusyMaxWaitMs) {
+            throw new QoderModelBusyError(
+              `Qoder server error: model is busy (10605 RESOURCE_LIMIT, requested delay ${requestedWaitMs}ms exceeds local cap ${modelBusyMaxWaitMs}ms)`,
+              requestedWaitMs,
+            );
+          }
+          await sleepWithAbort(requestedWaitMs, options?.signal);
+          // Covers an abort that lands between the wait and the next attempt.
+          options?.signal?.throwIfAborted();
+          // qodercli resets request_id and chat_record_id to ONE new value per
+          // recovery attempt and keeps request_set_id, so the server can
+          // correlate the retry with the same request set.
+          const nextId = crypto.randomUUID();
+          reqBody.request_id = nextId;
+          reqBody.chat_record_id = nextId;
+          encodedBytes = qoderEncodeBody(Buffer.from(JSON.stringify(reqBody)));
+          headers = buildAuthHeaders(encodedBytes, chatURL, {
+            userID,
+            authToken: accessToken,
+            name,
+            email,
+            machineID,
+          });
+        }
+      }
     } catch (e: unknown) {
       output.stopReason = options?.signal?.aborted ? "aborted" : "error";
       output.errorMessage = e instanceof Error ? e.message : String(e);