From c9ffe4adfa2ad20653872d29b0e675371381fd12 Mon Sep 17 00:00:00 2001 From: caiji <522caiji@gmail.com> Date: Tue, 8 Sep 2026 20:39:13 +0800 Subject: [PATCH] fix: tolerate completed Qoder streams without done --- deploy/README.md | 5 ++- deploy/docker-compose.yml | 3 +- worker/src/sse.mjs | 78 ++++++++++++++++++++++++++++++++++++++- worker/test/sse.test.mjs | 28 +++++++++++++- 4 files changed, 109 insertions(+), 5 deletions(-) diff --git a/deploy/README.md b/deploy/README.md index 6a21cd6..50966aa 100644 --- a/deploy/README.md +++ b/deploy/README.md @@ -97,6 +97,7 @@ the same conversation prefer the same account from the first user message | `QODER_DATA_DIR` | `/data` | SQLite database and durable account credentials | | `QODER_RUNTIME_DIR` | `/run/cli2api` | Ephemeral per-account runtime homes for providers that use child processes | | `QODER_MAX_RETRY_ACCOUNTS` | `4` | Maximum accounts attempted for one request (1-64) | +| `QODER_SSE_DIAGNOSTIC_MODELS` | empty | Comma-separated Qoder model IDs for redacted SSE diagnostics in Runtime Logs; `*` enables all | | `QODER_WORKER_BASE_PORT` | `32100` | Internal child-runtime port range | | `QODERCLI_JS` | image default | Pinned Qoder Global CLI bundle | | `QODERCNCLI_JS` | image default | Pinned Qoder CN CLI bundle | @@ -105,6 +106,8 @@ the same conversation prefer the same account from the first user message | `UPDATE_AGENT_TOKEN` | empty | Docker Desktop updater token, written by the installer | | `CLI2API_UPDATER_SOCKET_DIR` | platform-specific | Host directory mounted read-only for the Linux updater socket | +For a problematic Qoder model, set `QODER_SSE_DIAGNOSTIC_MODELS` to the exact model ID shown by `/v1/models`, for example `qoder-extreme`, then recreate the container. Diagnostics record only SSE shape, event counts, status codes, finish signals, and field lengths; they do not record prompt, response, tool arguments, or token contents. + Per-account concurrency is configured through `max_inflight` in the console; it is persisted with each account and passed to its runtime. @@ -202,4 +205,4 @@ accepted for staged-update hosts; unknown versions fail closed. - Published host updater assets cover Linux, macOS, and Windows on `amd64` and `arm64`. - Run the Windows updater installer as the same signed-in user that runs Docker Desktop; do not install it as `LocalSystem`. - Keep `deploy/.env` private because Docker Desktop mode stores the updater token there. -- Do not remove `qoder-data` unless you intentionally want to delete SQLite accounts and credentials. +- Do not remove `qoder-data` unless you intentionally want to delete SQLite accounts and credentials. \ No newline at end of file diff --git a/deploy/docker-compose.yml b/deploy/docker-compose.yml index 5d12764..43c6eba 100644 --- a/deploy/docker-compose.yml +++ b/deploy/docker-compose.yml @@ -10,6 +10,7 @@ services: PORT: "3010" QODER_DATA_DIR: "/data" QODER_RUNTIME_DIR: "/run/cli2api" + QODER_SSE_DIAGNOSTIC_MODELS: "${QODER_SSE_DIAGNOSTIC_MODELS:-}" QODER_WORKER_BASE_PORT: "${QODER_WORKER_BASE_PORT:-32100}" UPDATE_SOCKET_PATH: "/run/cli2api-updater/updater.sock" UPDATE_AGENT_URL: "${UPDATE_AGENT_URL:-}" @@ -31,4 +32,4 @@ services: restart: unless-stopped volumes: - qoder-data: + qoder-data: \ No newline at end of file diff --git a/worker/src/sse.mjs b/worker/src/sse.mjs index 68ad602..c2d1dc0 100644 --- a/worker/src/sse.mjs +++ b/worker/src/sse.mjs @@ -2,6 +2,32 @@ import { estimateTokens } from "./plaintext.mjs"; import { resolveUsage, usageLooksUseful } from "./usage.mjs"; import { classifyError } from "./errors.mjs"; +const diagnosticModels = new Set( + String(process.env.QODER_SSE_DIAGNOSTIC_MODELS || "") + .split(",") + .map((model) => model.trim().toLowerCase().replace(/[\s_]+/g, "-")) + .filter(Boolean), +); + +function shouldDiagnoseModel(model) { + const normalized = String(model || "").trim().toLowerCase().replace(/[\s_]+/g, "-"); + return diagnosticModels.has("*") || diagnosticModels.has(normalized); +} + +function summarizeSseBody(body) { + if (!body || typeof body !== "object") return { kind: typeof body }; + const choice = Array.isArray(body.choices) ? body.choices[0] : null; + const delta = choice?.delta && typeof choice.delta === "object" ? choice.delta : null; + return { + keys: Object.keys(body).slice(0, 24), + choiceKeys: choice ? Object.keys(choice).slice(0, 24) : [], + deltaKeys: delta ? Object.keys(delta).slice(0, 24) : [], + choiceCount: Array.isArray(body.choices) ? body.choices.length : 0, + hasError: Boolean(body.error), + hasUsage: usageLooksUseful(body), + }; +} + function scoreUpstreamError(err) { if (!err || typeof err !== "object") return 0; const code = String(err.code || "").toLowerCase(); @@ -198,6 +224,39 @@ export async function pipeNestedSseToOpenAI(upstreamRes, res, { model = "auto", let nextToolCallIndex = 0; let finishReason = null; let usageAcc = null; + const diagnostic = shouldDiagnoseModel(model); + let frameCount = 0; + let parseErrorCount = 0; + let usageSeen = false; + const eventNames = new Set(); + const frameKinds = new Set(); + const statusCodes = new Set(); + const bodyShapes = []; + const recordBodyShape = (body) => { + if (!diagnostic || bodyShapes.length >= 12) return; + bodyShapes.push(summarizeSseBody(body)); + }; + const emitDiagnostic = (stage) => { + if (!diagnostic) return; + console.error("[sse] qoder model diagnostic", JSON.stringify({ + stage, + model, + upstreamStatus: upstreamRes.status ?? null, + eventCount, + frameCount, + eventNames: [...eventNames], + frameKinds: [...frameKinds], + statusCodes: [...statusCodes], + bodyShapes, + sawDone, + finishReason, + contentLength: content.length, + reasoningLength: reasoning.length, + toolCallCount: toolCallsByIndex.size, + parseErrorCount, + usageSeen, + })); + }; const writeChunk = (delta, finish_reason = null) => { res.write( @@ -226,6 +285,7 @@ export async function pipeNestedSseToOpenAI(upstreamRes, res, { model = "auto", readResult = await reader.read(); } catch (err) { const detail = err?.message || String(err); + emitDiagnostic("read_error"); console.error("[sse] upstream stream read failed", { model, eventCount, @@ -265,11 +325,15 @@ export async function pipeNestedSseToOpenAI(upstreamRes, res, { model = "auto", if (!line.startsWith("data:")) continue; const raw = line.slice(5).trim(); if (!raw) continue; + frameCount += 1; + if (eventName) eventNames.add(eventName); try { if (eventName === "error") { try { const errBody = JSON.parse(raw); sawError = rememberError(sawError, errBody.error || errBody); + frameKinds.add("error"); + recordBodyShape(errBody); eventCount += 1; } catch { sawError = rememberError(sawError, { message: raw }); @@ -280,12 +344,17 @@ export async function pipeNestedSseToOpenAI(upstreamRes, res, { model = "auto", const extracted = extractDeltaFromOuter(raw); if (extracted.done) { sawDone = true; + frameKinds.add("done"); continue; } const body = extracted.body; + frameKinds.add("body"); + if (extracted.statusCode != null) statusCodes.add(Number(extracted.statusCode)); + recordBodyShape(body); if (!body) continue; eventCount += 1; if (usageLooksUseful(body)) { + usageSeen = true; usageAcc = resolveUsage([usageAcc, body], usageAcc || {}); } // Nested OpenAI-style error: { error: { message, type, code } } @@ -360,6 +429,8 @@ export async function pipeNestedSseToOpenAI(upstreamRes, res, { model = "auto", } eventName = "message"; } catch (err) { + parseErrorCount += 1; + frameKinds.add("invalid"); console.error("[sse] frame parse failed", { model, eventCount, @@ -371,6 +442,7 @@ export async function pipeNestedSseToOpenAI(upstreamRes, res, { model = "auto", } if (sawError) { + emitDiagnostic("provider_error"); console.error("[sse] upstream provider error", { model, eventCount, @@ -389,7 +461,8 @@ export async function pipeNestedSseToOpenAI(upstreamRes, res, { model = "auto", ); } - if (!sawDone) { + if (!sawDone && !finishReason) { + emitDiagnostic("incomplete"); console.error("[sse] upstream stream ended without done", { model, eventCount, @@ -409,6 +482,7 @@ export async function pipeNestedSseToOpenAI(upstreamRes, res, { model = "auto", throw new Error("upstream_stream_incomplete: upstream stream ended before [DONE]"); } + emitDiagnostic("complete"); const usage = resolveUsage(usageAcc, { prompt_tokens: promptTokens, completion_tokens: estimatedCompletion || estimateTokens(content + reasoning), @@ -472,4 +546,4 @@ export async function pipeNestedSseToOpenAI(upstreamRes, res, { model = "auto", tool_calls, finish_reason: finalReason, }; -} +} \ No newline at end of file diff --git a/worker/test/sse.test.mjs b/worker/test/sse.test.mjs index 00f838f..1d827ed 100644 --- a/worker/test/sse.test.mjs +++ b/worker/test/sse.test.mjs @@ -110,6 +110,32 @@ test("reports a structured error when upstream ends without DONE", async () => { }); +test("accepts a completed choice when upstream omits DONE", async () => { + const upstream = { + body: new ReadableStream({ + start(controller) { + const body = JSON.stringify({ + body: JSON.stringify({ + choices: [ + { delta: { content: "complete" }, finish_reason: "stop" }, + ], + }), + }); + controller.enqueue(new TextEncoder().encode("data: " + body + "\n\n")); + controller.close(); + }, + }), + }; + let output = ""; + const res = { write(chunk) { output += chunk; } }; + + const result = await pipeNestedSseToOpenAI(upstream, res, { model: "qoder-extreme" }); + assert.equal(result.content, "complete"); + assert.match(output, /"finish_reason":"stop"/); + assert.match(output, /data: \[DONE\]/); + assert.doesNotMatch(output, /upstream_stream_incomplete/); +}); + test("normalizes missing tool call indexes and ids", async () => { const nested = (body) => "data: " + JSON.stringify({ body: JSON.stringify(body) }) + "\n"; const upstream = { @@ -134,4 +160,4 @@ test("normalizes missing tool call indexes and ids", async () => { assert.ok(calls[0].id); assert.ok(calls[1].id); assert.notEqual(calls[0].id, calls[1].id); -}); +}); \ No newline at end of file