From 48ab32e28557b96c82ddfcd07cd1dc2b12eaa771 Mon Sep 17 00:00:00 2001 From: summer-tt Date: Thu, 22 Oct 2026 20:22:19 +0800 Subject: [PATCH] fix(stream): recover from transient model-busy rejections Recognize Qoder's nested 10605 scheduler response and retry only ready rejections before any model output. Bound retries and delays, honor aborts, refresh request IDs and signatures, and clear all per-attempt metadata and usage. Surface a retryable error when local recovery is exhausted, while rejecting retries after partial text, thinking, or tool output. Add synthetic regression coverage including pi-ai retry integration. --- src/__tests__/model-busy-retry.test.ts | 570 +++++++++++++++ src/protocol/stream.ts | 922 +++++++++++++++---------- 2 files changed, 1127 insertions(+), 365 deletions(-) create mode 100644 src/__tests__/model-busy-retry.test.ts 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[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 { + return { id: "Ultimate", api: "qoder-api" as Api, provider: "qoder" } as Model; +} + +function makeContext(): Context { + return { + systemPrompt: "test", + messages: [{ role: "user", content: "hi" }], + tools: [], + } as unknown as Context; +} + +async function consume(stream: AssistantMessageEventStream): Promise { + 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 { + 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; +} + +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)["Cosy-Bodyhash"]; + const secondHash = (secondInit.headers as Record)["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: " { + 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; + + 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; + + 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 { + 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 | 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) + : 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 { + 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; complete: Set; @@ -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 => { + // 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 `` - // opener into reasoning_content (with the matching - // `` 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 `` + // opener into reasoning_content (with the matching + // `` 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, + 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, - 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);