fix: prevent non-SSE stream pipe crash and cross-IdP account overwrites (#2244)

- streamingHandler: when upstream returns non-SSE/JSON (e.g. Cloudflare
  5xx HTML), read body, sanitize <title>, notify streamController and
  return a clean JSON error instead of crashing the pipe.
- connectionsRepo: dedup OAuth connections on (email + username) so
  cross-IdP accounts sharing an email no longer overwrite each other;
  workspace providers keep workspace-id matching.
- kimchi: bump User-Agent to 0.1.50, add svg asset + browser-login
  service, and 21 unit tests.

Co-authored-by: Cursor <cursoragent@cursor.com>

open-sse/handlers/chatCore/streamingHandler.js

@@ -43,7 +43,7 @@ function buildTransformStream({ provider, sourceFormat, targetFormat, userAgent,
 /**
  * Handle streaming response — pipe provider SSE through transform stream to client.
  */
-export function handleStreamingResponse({ providerResponse, provider, model, sourceFormat, targetFormat, userAgent, body, stream, translatedBody, finalBody, requestStartTime, connectionId, apiKey, clientRawRequest, onRequestSuccess, reqLogger, toolNameMap, streamController, onStreamComplete }) {
+export async function handleStreamingResponse({ providerResponse, provider, model, sourceFormat, targetFormat, userAgent, body, stream, translatedBody, finalBody, requestStartTime, connectionId, apiKey, clientRawRequest, onRequestSuccess, reqLogger, toolNameMap, streamController, onStreamComplete }) {
   if (onRequestSuccess) {
     Promise.resolve()
       .then(onRequestSuccess)
@@ -52,12 +52,30 @@ export function handleStreamingResponse({ providerResponse, provider, model, sou
       });
   }
 
-  // Warn when upstream returns unexpected Content-Type for a streaming response.
-  // This often means the provider returned an HTML error page or plain-text error
-  // that the SSE transform stream would forward as garbage to the client.
+  // When upstream returns HTML/text instead of SSE (e.g. Cloudflare 5xx error
+  // page), piping it through the SSE transform stream causes Next.js
+  // "failed to pipe response" and crashes the chat router. Read the body,
+  // pull a short human-readable message from the <title>, sanitize it, and
+  // return a clean JSON error instead. The message is stripped of HTML tags
+  // and clamped so untrusted upstream text never reaches the client verbatim
+  // (the UI may render error.message as HTML).
   const upstreamContentType = (providerResponse.headers.get('content-type') || '').toLowerCase();
   if (upstreamContentType && !upstreamContentType.includes('text/event-stream') && !upstreamContentType.includes('application/json')) {
-    console.warn('[STREAM] ' + provider + ' | ' + model + ' | unexpected Content-Type: ' + upstreamContentType);
+    const bodyText = await providerResponse.text().catch(() => '');
+    const titleMatch = bodyText.match(/<title>([^<]+)<\/title>/i);
+    const sanitizedTitle = (titleMatch?.[1] || '').replace(/<[^>]*>/g, '').replace(/[\r\n]+/g, ' ').trim().slice(0, 160);
+    const shortMsg = sanitizedTitle
+      || (bodyText.length < 200 ? bodyText.replace(/<[^>]*>/g, '').trim().slice(0, 160) : `Upstream returned non-SSE response (${upstreamContentType})`);
+    const status = providerResponse.status || 502;
+    console.warn(`[STREAM] ${provider} | ${model} | blocked pipe: ${shortMsg} [${status}]`);
+    streamController?.handleError?.(new Error(`upstream non-SSE: ${status}`));
+    return {
+      success: false,
+      response: new Response(JSON.stringify({ error: { message: `[${status}]: ${shortMsg}` } }), {
+        status,
+        headers: { 'Content-Type': 'application/json', 'Access-Control-Allow-Origin': '*' },
+      }),
+    };
   }
 
   const transformStream = buildTransformStream({ provider, sourceFormat, targetFormat, userAgent, reqLogger, toolNameMap, model, connectionId, body, onStreamComplete, apiKey });

open-sse/providers/registry/kimchi.js

@@ -20,7 +20,7 @@ export default {
     baseUrl: "https://llm.kimchi.dev/openai/v1/chat/completions",
     format: "openai",
     headers: {
-      "User-Agent": "kimchi/0.1.40",
+      "User-Agent": "kimchi/0.1.50",
     },
     auth: {
       combined: true,

src/lib/db/repos/connectionsRepo.js

@@ -98,13 +98,26 @@ export async function createProviderConnection(data) {
 
     let existing = null;
     if (data.authType === "oauth" && data.email) {
+      const incomingUsername = data.providerSpecificData?.username;
       const incomingWs = data.providerSpecificData?.chatgptAccountId;
       existing = all.find(c => {
         if (c.authType !== "oauth" || c.email !== data.email) return false;
-        // If both sides have a workspace ID, they must match for dedup
+        // Workspace providers (Codex) use workspace ID when both sides have it
         const existingWs = c.providerSpecificData?.chatgptAccountId;
         if (incomingWs && existingWs) return incomingWs === existingWs;
-        return true; // fallback: email-only match for non-workspace providers
+        if (incomingWs && !existingWs) return false;
+        if (!incomingWs && existingWs) return false;
+        // Non-workspace providers: match on (email + username) so cross-IdP
+        // accounts don't overwrite each other. Require username on both sides
+        // — if only one side has it, treat as a distinct identity rather than
+        // collapsing onto the bare-email fallback (which would re-introduce
+        // the cross-IdP overwrite).
+        const existingUsername = c.providerSpecificData?.username;
+        if (incomingUsername && existingUsername) {
+          return incomingUsername === existingUsername;
+        }
+        if (incomingUsername || existingUsername) return false;
+        return true;
       });
     } else if (data.authType === "apikey" && data.name) {
       existing = all.find(c => c.authType === "apikey" && c.name === data.name);

src/lib/oauth/services/kimchi.js

@@ -0,0 +1,133 @@
+// Kimchi browser-login service.
+//
+// Ports Kimchi CLI's authenticateViaBrowser (src/cli-auth/index.ts) onto
+// 9Router's shared startLocalServer util (same one xai/antigravity use).
+// Simpler than those: the token arrives directly on the callback query
+// string — no authorization-code exchange, no PKCE.
+//
+// In-flight logins are held in `sessions` keyed by state. The OAuthModal
+// device_code flow starts one via requestDeviceCode(); pollToken() peeks
+// at the resolved token; the generic [provider]/[action] route calls
+// createProviderConnection with the real token.
+import { randomBytes } from "node:crypto";
+import { startLocalServer } from "../utils/server.js";
+import { KIMCHI_CONFIG } from "../constants/oauth.js";
+
+const sessions = new Map(); // state -> { result, close, timeout, done, resolved }
+const SESSION_TTL_MS = 5 * 60 * 1000;
+
+export function buildKimchiAuthUrl(callbackUrl, state) {
+  const params = new URLSearchParams({ callback: callbackUrl, state });
+  return `${KIMCHI_CONFIG.webAppUrl}/cli-auth?${params.toString()}`;
+}
+
+export function generateState() {
+  return randomBytes(32).toString("hex");
+}
+
+// Returns the resolved { token } if the session for `state` has completed,
+// or null if it is still pending / unknown.
+export function getResolvedSession(state) {
+  const s = sessions.get(state);
+  if (!s || !s.done || !s.resolved) return null;
+  return s.resolved;
+}
+
+export class KimchiService {
+  async startLogin() {
+    const state = generateState();
+    let resolveResult;
+    const result = new Promise((resolve) => { resolveResult = resolve; });
+
+    const { port, close } = await startLocalServer((params) => {
+      this._handleCallback(params, state)
+        .then(resolveResult)
+        .catch((err) => resolveResult({ error: err.message }));
+    });
+
+    const timeout = setTimeout(() => {
+      resolveResult({ error: "Browser login timed out — please try again" });
+      close();
+    }, KIMCHI_CONFIG.callbackTimeoutMs);
+
+    sessions.set(state, { result, close, timeout, done: false, resolved: null });
+
+    // Stash the resolved value so pollToken() can retrieve the real token,
+    // close the loopback server, and reap the session after a TTL so the
+    // Map can't grow unbounded across many logins.
+    result.then((r) => {
+      const s = sessions.get(state);
+      if (!s) return;
+      s.done = true;
+      s.resolved = r;
+      clearTimeout(s.timeout);
+      try { s.close(); } catch { /* already closed */ }
+      setTimeout(() => sessions.delete(state), SESSION_TTL_MS).unref?.();
+    });
+
+    const callbackUrl = `http://127.0.0.1:${port}${KIMCHI_CONFIG.callbackPath}`;
+    const authUrl = buildKimchiAuthUrl(callbackUrl, state);
+    return { authUrl, port, state, result, close };
+  }
+
+  async _handleCallback(params, expectedState) {
+    if (params.error) {
+      throw new Error(params.error_description || params.error);
+    }
+    const candidate = params.state;
+    if (!candidate || candidate !== expectedState) {
+      throw new Error("This request isn't valid. Please restart the Kimchi login flow.");
+    }
+    const token = params.token;
+    if (!token) {
+      throw new Error("No token was returned by the Kimchi authentication server");
+    }
+    const check = await this.validateToken(token);
+    if (!check.valid) {
+      throw new Error(check.error || "Kimchi token validation failed");
+    }
+    return { token };
+  }
+
+  async fetchProfile(token) {
+    try {
+      const res = await fetch(KIMCHI_CONFIG.meUrl, {
+        headers: { Authorization: `Bearer ${token}` },
+      });
+      if (!res.ok) return {};
+      const j = await res.json();
+      return { displayName: j.name, email: j.email, username: j.username };
+    } catch {
+      return {};
+    }
+  }
+
+  // Validate a token against Kimchi's supported-providers endpoint.
+  // 200 → valid; 401/403 → invalid; anything else (incl. network/timeout)
+  // → fail-open valid so a flaky validation never blocks a good login.
+  async validateToken(token) {
+    const controller = new AbortController();
+    const timer = setTimeout(() => controller.abort(), 10_000);
+    let status = 0;
+    try {
+      const res = await fetch(KIMCHI_CONFIG.validationUrl, {
+        method: "GET",
+        headers: {
+          Authorization: `Bearer ${token}`,
+          Accept: "application/json",
+        },
+        signal: controller.signal,
+      });
+      status = res.status;
+    } catch {
+      // Network error / abort → fail-open
+      return { valid: true };
+    } finally {
+      clearTimeout(timer);
+    }
+    if (status === 200) return { valid: true };
+    if (status === 401) return { valid: false, error: "Kimchi token invalid or expired" };
+    if (status === 403) return { valid: false, error: "Kimchi token lacks required scope" };
+    return { valid: true };
+  }
+}

tests/unit/kimchi.test.js

@@ -0,0 +1,234 @@
+import { describe, it, before } from "node:test";
+import assert from "node:assert/strict";
+
+// Load the registry entry once for the suite so a load failure is reported
+// next to the failing test instead of cascading as "undefined" in every
+// later assertion.
+let kimchiEntry;
+
+describe("kimchi registry entry", () => {
+  before(async () => {
+    kimchiEntry = (await import("../../open-sse/providers/registry/kimchi.js")).default;
+  });
+
+  it("is an oauth provider auto-listed via byCategory", () => {
+    assert.equal(kimchiEntry.id, "kimchi");
+    assert.equal(kimchiEntry.category, "oauth");
+  });
+
+  it("points at the OpenAI-compatible gateway with an authenticated UA", () => {
+    assert.equal(
+      kimchiEntry.transport.baseUrl,
+      "https://llm.kimchi.dev/openai/v1/chat/completions",
+    );
+    // UA must be a non-empty string the gateway can identify; the value
+    // itself is owned by the Kimchi CLI release and may change upstream.
+    const ua = kimchiEntry.transport.headers["User-Agent"];
+    assert.ok(typeof ua === "string" && ua.length > 0, `User-Agent missing: ${ua}`);
+  });
+
+  it("uses Bearer auth", () => {
+    assert.deepEqual(kimchiEntry.transport.auth, {
+      combined: true,
+      header: "Authorization",
+      scheme: "bearer",
+    });
+  });
+
+  it("exposes the upstream static models", () => {
+    const ids = kimchiEntry.models.map((m) => m.id);
+    assert.ok(ids.includes("kimi-k2.7"));
+    assert.ok(ids.includes("minimax-m3"));
+    assert.ok(ids.includes("nemotron-3-ultra-fp4"));
+    assert.ok(ids.length >= 5, `expected >= 5 static models, got ${ids.length}`);
+  });
+
+  it("passes through models not in the static list", () => {
+    assert.equal(kimchiEntry.passthroughModels, true);
+  });
+});
+
+// ── Pure-function clones of the service logic (tested in isolation so
+//     node --test works without resolving the Next.js Webpack "open-sse"
+//     alias that src/lib/oauth/services/kimchi.js's dependency imports). ──
+
+function buildKimchiAuthUrl(callbackUrl, state) {
+  const params = new URLSearchParams({ callback: callbackUrl, state });
+  return `https://app.kimchi.dev/cli-auth?${params.toString()}`;
+}
+
+async function _handleCallback(params, expectedState) {
+  if (params.error) {
+    throw new Error(params.error_description || params.error);
+  }
+  const candidate = params.state;
+  if (!candidate || candidate !== expectedState) {
+    throw new Error(
+      "This request isn't valid. Please restart the Kimchi login flow.",
+    );
+  }
+  const token = params.token;
+  if (!token) {
+    throw new Error("No token was returned by the Kimchi authentication server");
+  }
+  return { token };
+}
+
+describe("kimchi oauth", () => {
+  it("builds the cli-auth URL with encoded callback + state", () => {
+    const url = buildKimchiAuthUrl("http://127.0.0.1:4321/callback", "abc123");
+    const parsed = new URL(url);
+    assert.equal(parsed.origin, "https://app.kimchi.dev");
+    assert.equal(parsed.pathname, "/cli-auth");
+    assert.equal(parsed.searchParams.get("callback"), "http://127.0.0.1:4321/callback");
+    assert.equal(parsed.searchParams.get("state"), "abc123");
+  });
+
+  it("rejects a callback whose state does not match", async () => {
+    await assert.rejects(
+      () => _handleCallback({ token: "castai_v1_x", state: "wrong" }, "expected"),
+      /restart/i,
+    );
+  });
+
+  it("accepts a callback with matching state and returns the token", async () => {
+    const res = await _handleCallback({ token: "castai_v1_x", state: "match" }, "match");
+    assert.equal(res.token, "castai_v1_x");
+  });
+});
+
+// ── kimchiModels service (pure mapping logic, tested in isolation) ──
+
+// Clone of the metadata→model mapper so node --test resolves without the
+// open-sse/Webpack alias chain the real module imports.
+function mapKimchiMetadata(raw) {
+  if (!Array.isArray(raw)) return [];
+  return raw.map((m) => ({
+    id: m.slug,
+    name: m.display_name || m.slug,
+    contextLength: m.limits?.context_window || null,
+    maxOutputTokens: m.limits?.max_output_tokens || null,
+    isReasoning: m.reasoning === true,
+  }));
+}
+
+describe("kimchiModels", () => {
+  it("maps Kimchi metadata entries to 9router model shape", () => {
+    const raw = [{
+      slug: "glm-5.2-fp8",
+      display_name: "GLM 5.2",
+      reasoning: true,
+      limits: { context_window: 1048576, max_output_tokens: 1048576 },
+    }];
+    const models = mapKimchiMetadata(raw);
+    assert.equal(models.length, 1);
+    assert.deepEqual(models[0], {
+      id: "glm-5.2-fp8",
+      name: "GLM 5.2",
+      contextLength: 1048576,
+      maxOutputTokens: 1048576,
+      isReasoning: true,
+    });
+  });
+
+  it("falls back to slug as name when display_name is empty", () => {
+    const models = mapKimchiMetadata([{ slug: "kimi-k2.7", display_name: "", reasoning: false, limits: {} }]);
+    assert.equal(models[0].name, "kimi-k2.7");
+    assert.equal(models[0].contextLength, null);
+    assert.equal(models[0].isReasoning, false);
+  });
+
+  it("returns empty array for non-array input", () => {
+    assert.deepEqual(mapKimchiMetadata(null), []);
+    assert.deepEqual(mapKimchiMetadata({}), []);
+  });
+});
+
+// ── validateToken logic (pure decision over a status code) ──
+
+// Mirrors the decision in KimchiService.validateToken without importing the
+// service (which pulls the open-sse Webpack alias chain).
+function decideValidity(status) {
+  if (status === 200) return { valid: true };
+  if (status === 401) return { valid: false, error: "Kimchi token invalid or expired" };
+  if (status === 403) return { valid: false, error: "Kimchi token lacks required scope" };
+  return { valid: true }; // fail-open on unknown / network error
+}
+
+describe("kimchi validateToken", () => {
+  it("200 → valid", () => {
+    assert.deepEqual(decideValidity(200), { valid: true });
+  });
+  it("401 → invalid, expired message", () => {
+    const r = decideValidity(401);
+    assert.equal(r.valid, false);
+    assert.match(r.error, /invalid or expired/i);
+  });
+  it("403 → invalid, scope message", () => {
+    const r = decideValidity(403);
+    assert.equal(r.valid, false);
+    assert.match(r.error, /scope/i);
+  });
+  it("unknown / network error → fail-open valid", () => {
+    assert.equal(decideValidity(500).valid, true);
+    assert.equal(decideValidity(0).valid, true);
+  });
+});
+
+// ── OAuth dedup logic (pure clone of connectionsRepo matcher) ──
+// Mimics the find() predicate in createProviderConnection for OAuth
+// connections, so we can test the IdP-collision fix in isolation.
+function findExistingOAuth(all, incoming) {
+  const incomingEmail = incoming.email;
+  const incomingUsername = incoming.providerSpecificData?.username;
+  const incomingWs = incoming.providerSpecificData?.chatgptAccountId;
+  return all.find((c) => {
+    if (c.authType !== "oauth" || c.email !== incomingEmail) return false;
+    const existingWs = c.providerSpecificData?.chatgptAccountId;
+    if (incomingWs && existingWs) return incomingWs === existingWs;
+    if (incomingWs && !existingWs) return false;
+    if (!incomingWs && existingWs) return false;
+    const existingUsername = c.providerSpecificData?.username;
+    if (incomingUsername && existingUsername) {
+      return incomingUsername === existingUsername;
+    }
+    if (incomingUsername || existingUsername) return false;
+    return true;
+  });
+}
+
+describe("kimchi OAuth dedup", () => {
+  const google = { authType: "oauth", email: "x@y.com", providerSpecificData: { username: "google-oauth2|123" } };
+  const hf = { authType: "oauth", email: "x@y.com", providerSpecificData: { username: "huggingface|456" } };
+  const legacy = { authType: "oauth", email: "x@y.com", providerSpecificData: {} };
+  const other = { authType: "oauth", email: "z@y.com", providerSpecificData: { username: "google-oauth2|789" } };
+
+  it("different email never matches", () => {
+    assert.equal(findExistingOAuth([other], google), undefined);
+  });
+
+  it("same email + same username = dedup (re-login same IdP)", () => {
+    const found = findExistingOAuth([google], { ...google });
+    assert.equal(found, google);
+  });
+
+  it("same email + different username = NO match (cross-IdP, the bug)", () => {
+    assert.equal(findExistingOAuth([google], hf), undefined);
+  });
+
+  it("legacy row without username matches incoming without username (backward compat)", () => {
+    assert.equal(findExistingOAuth([legacy], { ...legacy }), legacy);
+  });
+
+  it("incoming without username does not match legacy row with username", () => {
+    assert.equal(findExistingOAuth([google], { ...legacy }), undefined);
+  });
+
+  it("workspaces still dedupe on workspace ID when both sides have one", () => {
+    const ws1 = { authType: "oauth", email: "a@b.com", providerSpecificData: { chatgptAccountId: "ws1" } };
+    const ws1dup = { authType: "oauth", email: "a@b.com", providerSpecificData: { chatgptAccountId: "ws1" } };
+    const ws2 = { authType: "oauth", email: "a@b.com", providerSpecificData: { chatgptAccountId: "ws2" } };
+    assert.equal(findExistingOAuth([ws1], ws1dup), ws1);
+    assert.equal(findExistingOAuth([ws1], ws2), undefined);
+  });
+});