diff --git a/open-sse/executors/qoder.js b/open-sse/executors/qoder.js index fa9bcbad..5e146fed 100644 --- a/open-sse/executors/qoder.js +++ b/open-sse/executors/qoder.js @@ -222,22 +222,75 @@ async function buildQoderRequestBody({ model, body, credentials, log, proxyOptio * and re-emit as `data: \n\n`. Errors become `data: [DONE]\n\n` plus * a synthetic OpenAI error chunk. */ -function wrapQoderSSE(response, model) { +async function wrapQoderSSE(response, model) { if (!response.ok || !response.body) return response; const decoder = new TextDecoder(); const encoder = new TextEncoder(); + + // Peek at first chunk to detect errors early + const reader = response.body.getReader(); + const firstRead = await reader.read(); + + if (firstRead.done) { + // Empty stream + return new Response("data: [DONE]\n\n", { + status: response.status, + statusText: response.statusText, + headers: { + "Content-Type": "text/event-stream", + "Cache-Control": "no-cache", + }, + }); + } + + // Parse first line to check for error + const firstText = decoder.decode(firstRead.value, { stream: true }); + const nlIndex = firstText.indexOf("\n"); + const firstLine = nlIndex !== -1 ? firstText.slice(0, nlIndex) : firstText; + const trimmed = firstLine.replace(/\r$/, "").trim(); + + if (trimmed.startsWith("data:")) { + const data = trimmed.slice(5).trimStart(); + if (data !== "[DONE]") { + try { + const envelope = JSON.parse(data); + const statusVal = typeof envelope.statusCodeValue === "number" ? envelope.statusCodeValue : 200; + + if (statusVal !== 200) { + // Error detected - return error Response to trigger failover + const msg = envelope.body || `upstream status ${statusVal}`; + const errorResponse = new Response( + JSON.stringify({ + error: { + message: `qoder error ${statusVal}: ${truncate(msg, 500)}`, + type: "upstream_error", + code: String(statusVal) + } + }), + { + status: statusVal >= 400 && statusVal < 600 ? statusVal : 502, + headers: { "Content-Type": "application/json" } + } + ); + reader.cancel(); + return errorResponse; + } + } catch (e) { + // Not JSON, continue as normal stream + } + } + } + + // No error detected - proceed with normal TransformStream let buffer = ""; let doneEmitted = false; - // Process one already-extracted SSE line (no trailing newline). Returns - // false when the line indicated end-of-stream so the caller can stop - // forwarding any remaining chunks after [DONE]. const processLine = (line, controller) => { const trimmed = line.replace(/\r$/, "").trim(); if (!trimmed) return; if (!trimmed.startsWith("data:")) return; - if (doneEmitted) return; // never forward chunks past stream end + if (doneEmitted) return; const data = trimmed.slice(5).trimStart(); if (data === "[DONE]") { @@ -270,10 +323,6 @@ function wrapQoderSSE(response, model) { doneEmitted = true; return; } - // Inner is an OpenAI-shaped chunk. Strip any embedded newlines so the - // SSE frame stays a single event (a literal "\n" inside `inner` would - // otherwise split the frame across multiple data: lines and downstream - // parsers would reassemble them as separate events). const sanitized = inner.replace(/\r?\n/g, ""); controller.enqueue(encoder.encode(`data: ${sanitized}\n\n`)); }; @@ -289,13 +338,7 @@ function wrapQoderSSE(response, model) { } }, flush(controller) { - // Finalize the decoder so any pending multi-byte sequence is - // released into `buffer` instead of being silently dropped. buffer += decoder.decode(); - // Drain any trailing line that arrived without a terminating newline - // (e.g. upstream closed the socket immediately after the last write, - // or a CDN stripped the final CRLF). Without this, the chunk that - // carries finish_reason is silently lost. if (buffer.length > 0) { processLine(buffer, controller); buffer = ""; @@ -307,9 +350,45 @@ function wrapQoderSSE(response, model) { }, }); - const transformed = response.body.pipeThrough(transform); - // Build a Response with passable headers; the streaming handler reads - // `.body` as a ReadableStream regardless of Content-Type. + // Create a ReadableStream that emits the first chunk + remaining chunks + const combinedStream = new ReadableStream({ + start(controller) { + controller.enqueue(firstRead.value); + }, + async pull(controller) { + const { done, value } = await reader.read(); + if (done) { + controller.close(); + } else { + controller.enqueue(value); + } + }, + cancel() { + reader.cancel(); + } + }); + + const transformed = combinedStream.pipeThrough(transform); + return new Response(transformed, { + status: response.status, + statusText: response.statusText, + headers: { + "Content-Type": "text/event-stream", + "Cache-Control": "no-cache", + }, + }); +} + controller.close(); + } else { + controller.enqueue(value); + } + }, + cancel() { + reader.cancel(); + } + }); + + const transformed = remainingStream.pipeThrough(transform); return new Response(transformed, { status: response.status, statusText: response.statusText, @@ -430,7 +509,7 @@ export class QoderExecutor extends BaseExecutor { return { response, url, headers, transformedBody: payload }; } - const wrapped = wrapQoderSSE(response, `qoder/${qoderKey}`); + const wrapped = await wrapQoderSSE(response, `qoder/${qoderKey}`); return { response: wrapped, url, headers, transformedBody: payload }; }