From bf7d4ebea5b37ba89fe8061d4d2074e3797e877a Mon Sep 17 00:00:00 2001 From: "R.D." Date: Thu, 2 Apr 2026 03:34:50 -0400 Subject: [PATCH] fix cc compatible streaming and compatible provider ui --- CHANGELOG.md | 4 + open-sse/handlers/chatCore.ts | 19 +- open-sse/handlers/sseParser.ts | 252 ++++++++++++++++++ open-sse/services/claudeCodeCompatible.ts | 2 +- .../(dashboard)/dashboard/providers/page.tsx | 54 ++-- src/lib/providers/validation.ts | 2 +- tests/unit/cc-compatible-provider.test.mjs | 88 ++++++ 7 files changed, 381 insertions(+), 40 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 2e7211c727..d32417d91e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,10 @@ ## [Unreleased] +### 🐛 Bug Fixes + +- **CC Compatible UX & Streaming:** Unified the Add CC/OpenAI/Anthropic compatible actions around the Anthropic UI treatment, and forced CC-compatible upstream requests to use SSE while still returning streaming or non-streaming responses based on the client request. + --- ## [3.4.3] - 2026-04-02 diff --git a/open-sse/handlers/chatCore.ts b/open-sse/handlers/chatCore.ts index 7a2887b641..15b91e191d 100644 --- a/open-sse/handlers/chatCore.ts +++ b/open-sse/handlers/chatCore.ts @@ -60,7 +60,11 @@ import { } from "../executors/codex.ts"; import { translateNonStreamingResponse } from "./responseTranslator.ts"; import { extractUsageFromResponse } from "./usageExtractor.ts"; -import { parseSSEToOpenAIResponse, parseSSEToResponsesOutput } from "./sseParser.ts"; +import { + parseSSEToClaudeResponse, + parseSSEToOpenAIResponse, + parseSSEToResponsesOutput, +} from "./sseParser.ts"; import { sanitizeOpenAIResponse } from "./responseSanitizer.ts"; import { withRateLimit, @@ -711,6 +715,7 @@ export async function handleChatCore({ let translatedBody = body; const isClaudePassthrough = sourceFormat === FORMATS.CLAUDE && targetFormat === FORMATS.CLAUDE; const isClaudeCodeCompatible = isClaudeCodeCompatibleProvider(provider); + const upstreamStream = stream || isClaudeCodeCompatible; let ccSessionId: string | null = null; // Determine if we should preserve client-side cache_control headers @@ -770,7 +775,7 @@ export async function handleChatCore({ sourceBody: body, normalizedBody: normalizedForCc, model, - stream, + stream: upstreamStream, sessionId: ccSessionId, cwd: process.cwd(), now: new Date(), @@ -1031,7 +1036,7 @@ export async function handleChatCore({ // Create stream controller for disconnect detection const streamController = createStreamController({ onDisconnect, log, provider, model }); - const dedupRequestBody = { ...translatedBody, model: `${provider}/${model}` }; + const dedupRequestBody = { ...translatedBody, model: `${provider}/${model}`, stream }; const dedupEnabled = shouldDeduplicate(dedupRequestBody); const dedupHash = dedupEnabled ? computeRequestHash(dedupRequestBody) : null; @@ -1065,7 +1070,7 @@ export async function handleChatCore({ const res = await executor.execute({ model: modelToCall, body: bodyToSend, - stream, + stream: upstreamStream, credentials: getExecutionCredentials(), signal: streamController.signal, log, @@ -1228,7 +1233,7 @@ export async function handleChatCore({ const retryResult = await executor.execute({ model: retryModelId, body: translatedBody, - stream, + stream: upstreamStream, credentials: getExecutionCredentials(), signal: streamController.signal, log, @@ -1532,7 +1537,9 @@ export async function handleChatCore({ const parsedFromSSE = targetFormat === FORMATS.OPENAI_RESPONSES ? parseSSEToResponsesOutput(rawBody, model) - : parseSSEToOpenAIResponse(rawBody, model); + : targetFormat === FORMATS.CLAUDE + ? parseSSEToClaudeResponse(rawBody, model) + : parseSSEToOpenAIResponse(rawBody, model); if (!parsedFromSSE) { appendRequestLog({ diff --git a/open-sse/handlers/sseParser.ts b/open-sse/handlers/sseParser.ts index 39758d409e..f21f780cc1 100644 --- a/open-sse/handlers/sseParser.ts +++ b/open-sse/handlers/sseParser.ts @@ -2,6 +2,75 @@ * Convert OpenAI-style SSE chunks into a single non-streaming JSON response. * Used as a fallback when upstream returns text/event-stream for stream=false. */ +function readSSEEvents(rawSSE) { + const lines = String(rawSSE || "").split("\n"); + const events = []; + let currentEvent = ""; + let currentData = []; + + const flush = () => { + if (currentData.length === 0) { + currentEvent = ""; + return; + } + + const payload = currentData.join("\n").trim(); + currentData = []; + if (!payload || payload === "[DONE]") { + currentEvent = ""; + return; + } + + try { + events.push({ + event: currentEvent || undefined, + data: JSON.parse(payload), + }); + } catch { + // Ignore malformed SSE events and continue best-effort parsing. + } + + currentEvent = ""; + }; + + for (const rawLine of lines) { + const line = rawLine.replace(/\r$/, ""); + if (line.trim() === "") { + flush(); + continue; + } + + if (line.startsWith("event:")) { + currentEvent = line.slice(6).trim(); + continue; + } + + if (line.startsWith("data:")) { + currentData.push(line.slice(5).trimStart()); + } + } + + flush(); + return events; +} + +function toRecord(value) { + return value && typeof value === "object" && !Array.isArray(value) ? value : {}; +} + +function toString(value, fallback = "") { + return typeof value === "string" ? value : fallback; +} + +function toNumber(value, fallback = 0) { + if (typeof value === "number" && Number.isFinite(value)) return value; + if (typeof value === "string" && value.trim().length > 0) { + const parsed = Number(value); + return Number.isFinite(parsed) ? parsed : fallback; + } + return fallback; +} + export function parseSSEToOpenAIResponse(rawSSE, fallbackModel) { const lines = String(rawSSE || "").split("\n"); const chunks = []; @@ -143,6 +212,189 @@ export function parseSSEToOpenAIResponse(rawSSE, fallbackModel) { return result; } +/** + * Convert Claude-style SSE events into a single non-streaming message object. + * Used when Claude-compatible upstreams stream even for stream=false. + */ +export function parseSSEToClaudeResponse(rawSSE, fallbackModel) { + const payloads = readSSEEvents(rawSSE) + .map((event) => toRecord(event.data)) + .filter((payload) => Object.keys(payload).length > 0); + + if (payloads.length === 0) return null; + + const blocks = new Map(); + const usage = {}; + let messageId = ""; + let model = fallbackModel || "claude"; + let role = "assistant"; + let stopReason = "end_turn"; + let stopSequence = null; + + const mergeUsage = (incoming) => { + const usageRecord = toRecord(incoming); + for (const [key, value] of Object.entries(usageRecord)) { + if (typeof value === "number" && Number.isFinite(value)) { + usage[key] = value; + } else if (value && typeof value === "object" && !Array.isArray(value)) { + usage[key] = { ...toRecord(usage[key]), ...toRecord(value) }; + } else if (typeof value === "string" && value.trim().length > 0) { + usage[key] = value; + } + } + }; + + const tryParseJson = (raw) => { + try { + return JSON.parse(raw); + } catch { + return raw; + } + }; + + for (const payload of payloads) { + const eventType = toString(payload.type); + if (eventType === "message_start") { + const message = toRecord(payload.message); + messageId = toString(message.id, messageId || `msg_${Date.now()}`); + model = toString(message.model, model); + role = toString(message.role, role); + mergeUsage(message.usage); + continue; + } + + if (eventType === "content_block_start") { + const index = toNumber(payload.index, blocks.size); + const contentBlock = toRecord(payload.content_block); + const blockType = toString(contentBlock.type); + + if (blockType === "thinking") { + blocks.set(index, { + type: "thinking", + index, + thinking: toString(contentBlock.thinking), + signature: + typeof contentBlock.signature === "string" ? contentBlock.signature : undefined, + }); + } else if (blockType === "tool_use") { + blocks.set(index, { + type: "tool_use", + index, + id: toString(contentBlock.id, `toolu_${Date.now()}_${index}`), + name: toString(contentBlock.name), + input: contentBlock.input ?? {}, + inputJson: "", + }); + } else { + blocks.set(index, { + type: "text", + index, + text: toString(contentBlock.text), + }); + } + continue; + } + + if (eventType === "content_block_delta") { + const index = toNumber(payload.index, 0); + const delta = toRecord(payload.delta); + const deltaType = toString(delta.type); + const existing = blocks.get(index); + + if (deltaType === "input_json_delta") { + const toolUse = + existing && existing.type === "tool_use" + ? existing + : { + type: "tool_use", + index, + id: `toolu_${Date.now()}_${index}`, + name: "", + input: {}, + inputJson: "", + }; + toolUse.inputJson += toString(delta.partial_json); + blocks.set(index, toolUse); + continue; + } + + if (deltaType === "thinking_delta" || typeof delta.thinking === "string") { + const thinking = + existing && existing.type === "thinking" + ? existing + : { type: "thinking", index, thinking: "", signature: undefined }; + thinking.thinking += toString(delta.thinking); + blocks.set(index, thinking); + continue; + } + + const textBlock = + existing && existing.type === "text" + ? existing + : { + type: "text", + index, + text: "", + }; + textBlock.text += toString(delta.text); + blocks.set(index, textBlock); + continue; + } + + if (eventType === "message_delta") { + const delta = toRecord(payload.delta); + stopReason = toString(delta.stop_reason, stopReason); + stopSequence = + typeof delta.stop_sequence === "string" ? String(delta.stop_sequence) : stopSequence; + mergeUsage(payload.usage); + continue; + } + + mergeUsage(payload.usage); + } + + const content = [...blocks.values()] + .sort((a, b) => a.index - b.index) + .flatMap((block) => { + if (block.type === "text") { + return block.text ? [{ type: "text", text: block.text }] : []; + } + if (block.type === "thinking") { + return block.thinking + ? [ + { + type: "thinking", + thinking: block.thinking, + ...(block.signature ? { signature: block.signature } : {}), + }, + ] + : []; + } + + const parsedInput = + block.inputJson.trim().length > 0 ? tryParseJson(block.inputJson) : block.input; + return [ + { + type: "tool_use", + id: block.id, + name: block.name, + input: parsedInput, + }, + ]; + }); + + return { + id: messageId || `msg_${Date.now()}`, + type: "message", + role, + model, + content, + stop_reason: stopReason, + ...(stopSequence ? { stop_sequence: stopSequence } : {}), + ...(Object.keys(usage).length > 0 ? { usage } : {}), + }; +} + /** * Convert Responses API SSE events into a single non-streaming response object. * Expects events such as response.created / response.in_progress / response.completed. diff --git a/open-sse/services/claudeCodeCompatible.ts b/open-sse/services/claudeCodeCompatible.ts index 24540d2612..baff923d87 100644 --- a/open-sse/services/claudeCodeCompatible.ts +++ b/open-sse/services/claudeCodeCompatible.ts @@ -116,7 +116,7 @@ export function buildClaudeCodeCompatibleValidationPayload(model = "claude-sonne max_tokens: 1, }, model, - stream: false, + stream: true, sessionId, cwd: process.cwd(), now: new Date(), diff --git a/src/app/(dashboard)/dashboard/providers/page.tsx b/src/app/(dashboard)/dashboard/providers/page.tsx index 48a9381d9f..2dee7dca97 100644 --- a/src/app/(dashboard)/dashboard/providers/page.tsx +++ b/src/app/(dashboard)/dashboard/providers/page.tsx @@ -518,25 +518,14 @@ export default function ProvidersPage() { )} {ccCompatibleProviderEnabled && ( - )} - @@ -1292,6 +1281,7 @@ AddAnthropicCompatibleModal.propTypes = { }; function AddCcCompatibleModal({ isOpen, onClose, onCreated }) { + const t = useTranslations("providers"); const [formData, setFormData] = useState({ name: "", prefix: "", @@ -1378,25 +1368,25 @@ function AddCcCompatibleModal({ isOpen, onClose, onCreated }) {
setFormData({ ...formData, name: e.target.value })} - placeholder="CC Compatible Production" - hint="Display name for this Claude Code-compatible provider" + placeholder={t("compatibleProdPlaceholder", { type: CC_COMPATIBLE_LABEL })} + hint={t("nameHint")} /> setFormData({ ...formData, prefix: e.target.value })} - placeholder="cc" - hint="Used for model aliases such as prefix/model-id" + placeholder="cc-prod" + hint={t("prefixHint")} /> setFormData({ ...formData, baseUrl: e.target.value })} - placeholder="https://example.com/v1" - hint="Base URL for the CC-compatible site. Do not include /messages." + placeholder="https://api.anthropic.com" + hint={t("compatibleBaseUrlHint", { type: CC_COMPATIBLE_LABEL })} /> {showAdvanced && (
setFormData({ ...formData, chatPath: e.target.value })} placeholder={CC_COMPATIBLE_DEFAULT_CHAT_PATH} - hint="Defaults to the strict Claude Code-compatible messages path" + hint={t("chatPathHint")} /> setFormData({ ...formData, modelsPath: e.target.value })} placeholder={CC_COMPATIBLE_DEFAULT_MODELS_PATH} - hint="Defaults to /models" + hint={t("modelsPathHint")} />
)}
setCheckKey(e.target.value)} @@ -1448,13 +1438,13 @@ function AddCcCompatibleModal({ isOpen, onClose, onCreated }) { disabled={!checkKey || validating || !formData.baseUrl.trim()} variant="secondary" > - {validating ? "Checking..." : "Check"} + {validating ? t("checking") : t("check")}
{validationResult && ( - {validationResult === "success" ? "Valid" : "Invalid"} + {validationResult === "success" ? t("valid") : t("invalid")} )}
@@ -1468,10 +1458,10 @@ function AddCcCompatibleModal({ isOpen, onClose, onCreated }) { submitting } > - {submitting ? "Creating..." : "Add"} + {submitting ? t("creating") : t("add")}
diff --git a/src/lib/providers/validation.ts b/src/lib/providers/validation.ts index 4f273e8204..288a4ff18a 100644 --- a/src/lib/providers/validation.ts +++ b/src/lib/providers/validation.ts @@ -616,7 +616,7 @@ export async function validateClaudeCodeCompatibleProvider({ try { const messagesRes = await fetch(joinClaudeCodeCompatibleUrl(baseUrl, chatPath), { method: "POST", - headers: buildClaudeCodeCompatibleHeaders(apiKey, false, sessionId), + headers: buildClaudeCodeCompatibleHeaders(apiKey, true, sessionId), body: JSON.stringify(payload), }); diff --git a/tests/unit/cc-compatible-provider.test.mjs b/tests/unit/cc-compatible-provider.test.mjs index 8035af38b5..23af6abb8f 100644 --- a/tests/unit/cc-compatible-provider.test.mjs +++ b/tests/unit/cc-compatible-provider.test.mjs @@ -16,6 +16,7 @@ const { CLAUDE_CODE_COMPATIBLE_DEFAULT_MODELS_PATH, joinClaudeCodeCompatibleUrl, } = await import("../../open-sse/services/claudeCodeCompatible.ts"); +const { handleChatCore } = await import("../../open-sse/handlers/chatCore.ts"); const { validateProviderApiKey } = await import("../../src/lib/providers/validation.ts"); const providerNodesRoute = await import("../../src/app/api/provider-nodes/route.ts"); const providerNodesValidateRoute = @@ -263,7 +264,94 @@ test("validateProviderApiKey uses CC skeleton request after /models fallback", a ); assert.equal(calls[1].body.model, "claude-sonnet-4-6"); assert.equal(calls[1].body.messages[0].role, "user"); + assert.equal(calls[1].body.stream, true); assert.equal(calls[1].headers["x-api-key"], "sk-test"); + assert.equal(calls[1].headers.Accept, "text/event-stream"); +}); + +test("handleChatCore forces upstream streaming for CC compatible while returning JSON to non-stream clients", async () => { + const calls = []; + globalThis.fetch = async (url, init = {}) => { + calls.push({ + url: String(url), + method: init.method || "GET", + headers: init.headers, + body: init.body ? JSON.parse(String(init.body)) : null, + }); + + return new Response( + [ + "event: message_start", + 'data: {"type":"message_start","message":{"id":"msg_1","type":"message","role":"assistant","model":"claude-sonnet-4-6","usage":{"input_tokens":7,"output_tokens":0}}}', + "", + "event: content_block_start", + 'data: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}', + "", + "event: content_block_delta", + 'data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"Hello from CC"}}', + "", + "event: message_delta", + 'data: {"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":5}}', + "", + "event: message_stop", + 'data: {"type":"message_stop"}', + "", + ].join("\n"), + { + status: 200, + headers: { + "content-type": "text/event-stream", + }, + } + ); + }; + + const result = await handleChatCore({ + body: { + model: "claude-sonnet-4-6", + messages: [{ role: "user", content: "Ping" }], + stream: false, + }, + modelInfo: { + provider: "anthropic-compatible-cc-test", + model: "claude-sonnet-4-6", + extendedContext: false, + }, + credentials: { + apiKey: "sk-test", + providerSpecificData: { + baseUrl: "https://proxy.example.com", + chatPath: CLAUDE_CODE_COMPATIBLE_DEFAULT_CHAT_PATH, + }, + }, + clientRawRequest: { + endpoint: "/v1/chat/completions", + body: { + model: "claude-sonnet-4-6", + messages: [{ role: "user", content: "Ping" }], + stream: false, + }, + headers: new Headers({ accept: "application/json" }), + }, + userAgent: "unit-test", + log: { + debug() {}, + info() {}, + warn() {}, + error() {}, + }, + }); + + assert.equal(result.success, true); + assert.equal(calls.length, 1); + assert.equal(calls[0].headers.Accept, "text/event-stream"); + assert.equal(calls[0].body.stream, true); + + const payload = await result.response.json(); + assert.equal(payload.choices[0].message.content, "Hello from CC"); + assert.equal(payload.choices[0].finish_reason, "stop"); + assert.equal(payload.usage.prompt_tokens, 2007); + assert.equal(payload.usage.completion_tokens, 5); }); test("provider-nodes create route rejects CC mode when feature flag is disabled", async () => {