pjandro/OmniRoute
1
0
Fork 0
mirror of https://github.com/diegosouzapw/OmniRoute.git synced 2026-07-31 04:12:10 +03:00
Code Issues Packages Projects Releases Wiki Activity

fix cc compatible streaming and compatible provider ui

Browse Source
This commit is contained in:
R.D.
2026-04-02 03:34:50 -04:00
parent e121fec599
commit bf7d4ebea5
7 changed files with 381 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
CHANGELOG.md
View File

@@ -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

19
open-sse/handlers/chatCore.ts
View File

@@ -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({

252
open-sse/handlers/sseParser.ts
View File

@@ -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.

2
open-sse/services/claudeCodeCompatible.ts
View File

@@ -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(),

54
src/app/(dashboard)/dashboard/providers/page.tsx
View File

@@ -518,25 +518,14 @@ export default function ProvidersPage() {
</button>
)}
{ccCompatibleProviderEnabled && (
<Button
size="sm"
variant="secondary"
icon="add"
onClick={() => setShowAddCcCompatibleModal(true)}
>
<Button size="sm" icon="add" onClick={() => setShowAddCcCompatibleModal(true)}>
{ADD_CC_COMPATIBLE_LABEL}
</Button>
)}
<Button size="sm" icon="add" onClick={() => setShowAddAnthropicCompatibleModal(true)}>
{t("addAnthropicCompatible")}
</Button>
<Button
size="sm"
variant="secondary"
icon="add"
onClick={() => setShowAddCompatibleModal(true)}
className="!bg-white !text-black hover:!bg-gray-100"
>
<Button size="sm" icon="add" onClick={() => setShowAddCompatibleModal(true)}>
{t("addOpenAICompatible")}
</Button>
</div>
@@ -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 }) {
<Modal isOpen={isOpen} title={ADD_CC_COMPATIBLE_LABEL} onClose={onClose}>
<div className="flex flex-col gap-4">
<Input
label="Name"
label={t("nameLabel")}
value={formData.name}
onChange={(e) => 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")}
/>
<Input
label="Prefix"
label={t("prefixLabel")}
value={formData.prefix}
onChange={(e) => setFormData({ ...formData, prefix: e.target.value })}
placeholder="cc"
hint="Used for model aliases such as prefix/model-id"
placeholder="cc-prod"
hint={t("prefixHint")}
/>
<Input
label="Base URL"
label={t("baseUrlLabel")}
value={formData.baseUrl}
onChange={(e) => 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 })}
/>
<button
type="button"
@@ -1411,7 +1401,7 @@ function AddCcCompatibleModal({ isOpen, onClose, onCreated }) {
>
▶
</span>
Advanced settings
{t("advancedSettings")}
</button>
{showAdvanced && (
<div
@@ -1419,24 +1409,24 @@ function AddCcCompatibleModal({ isOpen, onClose, onCreated }) {
className="flex flex-col gap-3 pl-2 border-l-2 border-border"
>
<Input
label="Chat Path"
label={t("chatPathLabel")}
value={formData.chatPath}
onChange={(e) => 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")}
/>
<Input
label="Models Path"
label={t("modelsPathLabel")}
value={formData.modelsPath}
onChange={(e) => setFormData({ ...formData, modelsPath: e.target.value })}
placeholder={CC_COMPATIBLE_DEFAULT_MODELS_PATH}
hint="Defaults to /models"
hint={t("modelsPathHint")}
/>
</div>
)}
<div className="flex gap-2">
<Input
label="API Key for Check"
label={t("apiKeyForCheck")}
type="password"
value={checkKey}
onChange={(e) => 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")}
</Button>
</div>
</div>
{validationResult && (
<Badge variant={validationResult === "success" ? "success" : "error"}>
{validationResult === "success" ? "Valid" : "Invalid"}
{validationResult === "success" ? t("valid") : t("invalid")}
</Badge>
)}
<div className="flex gap-2">
@@ -1468,10 +1458,10 @@ function AddCcCompatibleModal({ isOpen, onClose, onCreated }) {
submitting
}
>
{submitting ? "Creating..." : "Add"}
{submitting ? t("creating") : t("add")}
</Button>
<Button onClick={onClose} variant="ghost" fullWidth>
Cancel
{t("cancel")}
</Button>
</div>
</div>

2
src/lib/providers/validation.ts
View File

@@ -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),
});

88
tests/unit/cc-compatible-provider.test.mjs
View File

@@ -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 () => {