fix(frontend): proxy LangGraph SSE without buffering

frontend/next.config.js

@@ -28,17 +28,6 @@ const config = {
       "http://127.0.0.1:8001",
     );
 
-    if (!process.env.NEXT_PUBLIC_LANGGRAPH_BASE_URL) {
-      rewrites.push({
-        source: "/api/langgraph",
-        destination: `${gatewayURL}/api`,
-      });
-      rewrites.push({
-        source: "/api/langgraph/:path*",
-        destination: `${gatewayURL}/api/:path*`,
-      });
-    }
-
     if (!process.env.NEXT_PUBLIC_BACKEND_BASE_URL) {
       rewrites.push({
         source: "/api/agents",
@@ -57,17 +46,31 @@ const config = {
         destination: `${gatewayURL}/api/skills/:path*`,
       });
 
-      // Catch-all for remaining gateway API routes (models, threads, memory,
-      // mcp, artifacts, uploads, suggestions, runs, etc.) that don't have
-      // their own NEXT_PUBLIC_* env var toggle.
-      //
-      // NOTE: this must come AFTER the /api/langgraph rewrite above so that
-      // LangGraph-compatible routes keep their public prefix while Gateway
-      // receives its native /api/* paths.
-      rewrites.push({
-        source: "/api/:path*",
-        destination: `${gatewayURL}/api/:path*`,
-      });
+      // Gateway API routes used outside the LangGraph SDK. Keep these
+      // explicit so /api/langgraph can be handled by the streaming route.
+      for (const path of [
+        "v1",
+        "assistants",
+        "channels",
+        "models",
+        "threads",
+        "runs",
+        "memory",
+        "mcp",
+        "artifacts",
+        "uploads",
+        "suggestions",
+        "feedback",
+      ]) {
+        rewrites.push({
+          source: `/api/${path}`,
+          destination: `${gatewayURL}/api/${path}`,
+        });
+        rewrites.push({
+          source: `/api/${path}/:path*`,
+          destination: `${gatewayURL}/api/${path}/:path*`,
+        });
+      }
     }
 
     return rewrites;

frontend/src/app/api/langgraph/[[...path]]/route.ts

@@ -0,0 +1,79 @@
+import { type NextRequest } from "next/server";
+
+export const runtime = "nodejs";
+export const dynamic = "force-dynamic";
+
+const GATEWAY_BASE_URL =
+  process.env.DEER_FLOW_INTERNAL_GATEWAY_BASE_URL ?? "http://127.0.0.1:8001";
+
+const HOP_BY_HOP_HEADERS = [
+  "connection",
+  "keep-alive",
+  "proxy-authenticate",
+  "proxy-authorization",
+  "te",
+  "trailer",
+  "transfer-encoding",
+  "upgrade",
+] as const;
+
+function buildGatewayUrl(path: string[] | undefined, search: string) {
+  const pathname = ["api", ...(path ?? []).map(encodeURIComponent)].join("/");
+  return `${GATEWAY_BASE_URL.replace(/\/+$/, "")}/${pathname}${search}`;
+}
+
+function buildHeaders(request: NextRequest) {
+  const headers = new Headers(request.headers);
+  for (const name of ["host", "content-length", ...HOP_BY_HOP_HEADERS]) {
+    headers.delete(name);
+  }
+  headers.set("accept-encoding", "identity");
+  return headers;
+}
+
+async function proxyRequest(
+  request: NextRequest,
+  context: { params: Promise<{ path?: string[] }> },
+) {
+  const { path } = await context.params;
+  const target = buildGatewayUrl(path, request.nextUrl.search);
+  const method = request.method.toUpperCase();
+  const hasBody = method !== "GET" && method !== "HEAD";
+  const init: RequestInit & { duplex?: "half" } = {
+    method,
+    headers: buildHeaders(request),
+    redirect: "manual",
+    cache: "no-store",
+    signal: request.signal,
+  };
+
+  if (hasBody) {
+    init.body = request.body;
+    init.duplex = "half";
+  }
+
+  const upstream = await fetch(target, init);
+  const headers = new Headers(upstream.headers);
+  for (const name of ["content-length", ...HOP_BY_HOP_HEADERS]) {
+    headers.delete(name);
+  }
+  const cacheControl = headers.get("Cache-Control");
+  headers.set(
+    "Cache-Control",
+    cacheControl ? `${cacheControl}, no-transform` : "no-cache, no-transform",
+  );
+  headers.set("X-Accel-Buffering", "no");
+
+  return new Response(upstream.body, {
+    status: upstream.status,
+    statusText: upstream.statusText,
+    headers,
+  });
+}
+
+export const GET = proxyRequest;
+export const POST = proxyRequest;
+export const PUT = proxyRequest;
+export const PATCH = proxyRequest;
+export const DELETE = proxyRequest;
+export const OPTIONS = proxyRequest;

frontend/tests/unit/app/api/langgraph-route.test.ts

@@ -0,0 +1,155 @@
+import { NextRequest } from "next/server";
+import { afterEach, beforeEach, describe, expect, test, vi } from "vitest";
+
+type RouteContext = {
+  params: Promise<{ path?: string[] }>;
+};
+
+async function loadRoute() {
+  vi.resetModules();
+  return await import("@/app/api/langgraph/[[...path]]/route");
+}
+
+function makeContext(path?: string[]): RouteContext {
+  return {
+    params: Promise.resolve({ path }),
+  };
+}
+
+describe("/api/langgraph route proxy", () => {
+  const originalGatewayBaseUrl =
+    process.env.DEER_FLOW_INTERNAL_GATEWAY_BASE_URL;
+
+  beforeEach(() => {
+    process.env.DEER_FLOW_INTERNAL_GATEWAY_BASE_URL =
+      "http://gateway.example/base/";
+  });
+
+  afterEach(() => {
+    vi.unstubAllGlobals();
+    if (originalGatewayBaseUrl === undefined) {
+      delete process.env.DEER_FLOW_INTERNAL_GATEWAY_BASE_URL;
+    } else {
+      process.env.DEER_FLOW_INTERNAL_GATEWAY_BASE_URL = originalGatewayBaseUrl;
+    }
+  });
+
+  test("re-encodes path segments and preserves query strings", async () => {
+    const fetchMock = vi.fn(async () => new Response("ok"));
+    vi.stubGlobal("fetch", fetchMock);
+
+    const { GET } = await loadRoute();
+    const request = new NextRequest(
+      "http://localhost:3000/api/langgraph/ignored?after=a%2Fb",
+    );
+    await GET(request, makeContext(["threads", "a/b", "x?y", "#z"]));
+
+    expect(fetchMock).toHaveBeenCalledWith(
+      "http://gateway.example/base/api/threads/a%2Fb/x%3Fy/%23z?after=a%2Fb",
+      expect.objectContaining({ method: "GET", signal: request.signal }),
+    );
+  });
+
+  test("strips proxy headers while forwarding auth headers and streamed body", async () => {
+    const fetchMock = vi.fn(async () => new Response("ok"));
+    vi.stubGlobal("fetch", fetchMock);
+
+    const { POST } = await loadRoute();
+    const request = new NextRequest(
+      "http://localhost:3000/api/langgraph/threads",
+      {
+        method: "POST",
+        headers: {
+          host: "localhost:3000",
+          connection: "keep-alive",
+          "content-length": "123",
+          "transfer-encoding": "chunked",
+          cookie: "access_token=abc",
+          "x-csrf-token": "csrf",
+          "accept-encoding": "gzip, br",
+        },
+        body: JSON.stringify({ ok: true }),
+      },
+    );
+    const body = request.body;
+
+    await POST(request, makeContext(["threads"]));
+
+    const fetchCalls = fetchMock.mock.calls as unknown as [
+      string,
+      RequestInit & { duplex?: "half" },
+    ][];
+    const init = fetchCalls[0]?.[1];
+    expect(init).toBeDefined();
+    expect(init?.method).toBe("POST");
+    expect(init?.body).toBe(body);
+    expect(init?.duplex).toBe("half");
+    const headers = init?.headers as Headers;
+    expect(headers.get("host")).toBeNull();
+    expect(headers.get("connection")).toBeNull();
+    expect(headers.get("content-length")).toBeNull();
+    expect(headers.get("transfer-encoding")).toBeNull();
+    expect(headers.get("cookie")).toBe("access_token=abc");
+    expect(headers.get("x-csrf-token")).toBe("csrf");
+    expect(headers.get("accept-encoding")).toBe("identity");
+  });
+
+  test("streams upstream bodies and normalizes response headers", async () => {
+    const stream = new ReadableStream({
+      start(controller) {
+        controller.enqueue(new TextEncoder().encode("event: metadata\n\n"));
+        controller.close();
+      },
+    });
+    const fetchMock = vi.fn(
+      async () =>
+        new Response(stream, {
+          status: 201,
+          statusText: "Created",
+          headers: {
+            "cache-control": "private, no-store",
+            "content-length": "999",
+            connection: "close",
+            "transfer-encoding": "chunked",
+          },
+        }),
+    );
+    vi.stubGlobal("fetch", fetchMock);
+
+    const { GET } = await loadRoute();
+    const response = await GET(
+      new NextRequest("http://localhost:3000/api/langgraph/threads"),
+      makeContext(["threads"]),
+    );
+
+    expect(response.status).toBe(201);
+    expect(response.statusText).toBe("Created");
+    expect(response.body).toBe(stream);
+    expect(response.headers.get("cache-control")).toBe(
+      "private, no-store, no-transform",
+    );
+    expect(response.headers.get("x-accel-buffering")).toBe("no");
+    expect(response.headers.get("content-length")).toBeNull();
+    expect(response.headers.get("connection")).toBeNull();
+    expect(response.headers.get("transfer-encoding")).toBeNull();
+  });
+
+  test("forwards OPTIONS requests", async () => {
+    const fetchMock = vi.fn(async () => new Response(null, { status: 204 }));
+    vi.stubGlobal("fetch", fetchMock);
+
+    const { OPTIONS } = await loadRoute();
+    const response = await OPTIONS(
+      new NextRequest("http://localhost:3000/api/langgraph/threads", {
+        method: "OPTIONS",
+      }),
+      makeContext(["threads"]),
+    );
+
+    expect(response.status).toBe(204);
+    expect(fetchMock).toHaveBeenCalledWith(
+      "http://gateway.example/base/api/threads",
+      expect.objectContaining({ method: "OPTIONS" }),
+    );
+  });
+});