feat(chat): add GET /api/chat/{chatId}/stream to resume an in-progress response

app/api/chat/[chatId]/stream/route.ts

@@ -0,0 +1,43 @@
+import type { NextRequest } from "next/server";
+import { NextResponse } from "next/server";
+import { getCorsHeaders } from "@/lib/networking/getCorsHeaders";
+import { handleResumeChatStream } from "@/lib/chat/handleResumeChatStream";
+
+// Matches POST /api/chat: a resumed stream stays open as long as the turn it
+// is following, so it needs the same ceiling rather than the route default.
+export const maxDuration = 800;
+export const dynamic = "force-dynamic";
+
+/**
+ * OPTIONS handler for CORS preflight requests.
+ *
+ * @returns A NextResponse with CORS headers.
+ */
+export async function OPTIONS() {
+  return new NextResponse(null, {
+    status: 200,
+    headers: getCorsHeaders(),
+  });
+}
+
+/**
+ * GET /api/chat/{chatId}/stream — reconnect to an in-progress chat response.
+ *
+ * The resume counterpart to `POST /api/chat`, which only resumes as a side
+ * effect of starting a turn. Pass `startIndex` to continue from the chunk
+ * after the last one received; omit it to read the response from the start.
+ *
+ * Contract: https://docs.recoupable.dev/api-reference/chat/workflow-stream
+ *
+ * @param request - The incoming NextRequest.
+ * @param options - Route options containing the async params.
+ * @param options.params - Route params containing the chat id.
+ * @returns A streaming 200, 204 when there is nothing to resume, or an error.
+ */
+export async function GET(
+  request: NextRequest,
+  options: { params: Promise<{ chatId: string }> },
+): Promise<Response> {
+  const { chatId } = await options.params;
+  return handleResumeChatStream(request, chatId);
+}

lib/chat/__tests__/handleResumeChatStream.test.ts

@@ -0,0 +1,166 @@
+import { describe, it, expect, vi, beforeEach } from "vitest";
+import { NextRequest, NextResponse } from "next/server";
+import { handleResumeChatStream } from "@/lib/chat/handleResumeChatStream";
+import { validateChatOwnership } from "@/lib/chat/validateChatOwnership";
+import { compareAndSetChatActiveStreamId } from "@/lib/chat/compareAndSetChatActiveStreamId";
+import { getRun } from "workflow/api";
+
+vi.mock("@/lib/chat/validateChatOwnership", () => ({ validateChatOwnership: vi.fn() }));
+vi.mock("@/lib/chat/compareAndSetChatActiveStreamId", () => ({
+  compareAndSetChatActiveStreamId: vi.fn(),
+}));
+vi.mock("workflow/api", () => ({ getRun: vi.fn() }));
+
+const CHAT_ID = "11111111-2222-3333-4444-555555555555";
+const RUN_ID = "wrun_01ABC";
+
+const request = (qs = "") =>
+  new NextRequest(`https://api.test/api/chat/${CHAT_ID}/stream${qs}`, { method: "GET" });
+
+/** Validator resolves with a chat carrying the given active_stream_id. */
+function withChat(activeStreamId: string | null) {
+  vi.mocked(validateChatOwnership).mockResolvedValue({
+    auth: { accountId: "acc-1" },
+    chat: { id: CHAT_ID, active_stream_id: activeStreamId },
+  } as never);
+}
+
+function withRun(
+  status: string,
+  getReadable = vi.fn(() => Object.assign(new ReadableStream(), { getTailIndex: async () => 41 })),
+) {
+  vi.mocked(getRun).mockReturnValue({
+    get status() {
+      return Promise.resolve(status);
+    },
+    getReadable,
+  } as never);
+  return getReadable;
+}
+
+beforeEach(() => {
+  vi.clearAllMocks();
+  vi.mocked(compareAndSetChatActiveStreamId).mockResolvedValue({
+    ok: true,
+    claimed: true,
+  } as never);
+});
+
+describe("handleResumeChatStream", () => {
+  it("returns 204 when the chat has no active stream", async () => {
+    withChat(null);
+
+    const res = await handleResumeChatStream(request(), CHAT_ID);
+
+    expect(res.status).toBe(204);
+    expect(getRun).not.toHaveBeenCalled();
+  });
+
+  it("returns 204 and clears the stale id when the run is already terminal", async () => {
+    withChat(RUN_ID);
+    withRun("completed");
+
+    const res = await handleResumeChatStream(request(), CHAT_ID);
+
+    expect(res.status).toBe(204);
+    expect(compareAndSetChatActiveStreamId).toHaveBeenCalledWith(CHAT_ID, RUN_ID, null);
+  });
+
+  it("streams the run and advertises the run id when the run is live", async () => {
+    withChat(RUN_ID);
+    withRun("running");
+
+    const res = await handleResumeChatStream(request(), CHAT_ID);
+
+    expect(res.status).toBe(200);
+    expect(res.headers.get("x-workflow-run-id")).toBe(RUN_ID);
+    expect(res.headers.get("content-type")).toContain("text/event-stream");
+  });
+
+  it("forwards startIndex to getReadable so a reconnect skips chunks already seen", async () => {
+    withChat(RUN_ID);
+    const getReadable = withRun("running");
+
+    await handleResumeChatStream(request("?startIndex=12"), CHAT_ID);
+
+    expect(getReadable).toHaveBeenCalledWith(expect.objectContaining({ startIndex: 12 }));
+  });
+
+  it("omits startIndex when absent so a fresh reader gets the whole turn", async () => {
+    withChat(RUN_ID);
+    const getReadable = withRun("running");
+
+    await handleResumeChatStream(request(), CHAT_ID);
+
+    expect(getReadable).toHaveBeenCalledWith(expect.objectContaining({ startIndex: undefined }));
+  });
+
+  it("returns 400 for a malformed startIndex without touching the run", async () => {
+    withChat(RUN_ID);
+    withRun("running");
+
+    const res = await handleResumeChatStream(request("?startIndex=-3"), CHAT_ID);
+
+    expect(res.status).toBe(400);
+    expect(getRun).not.toHaveBeenCalled();
+  });
+
+  it("propagates the validator's response (401/403/404) unchanged", async () => {
+    vi.mocked(validateChatOwnership).mockResolvedValue(
+      NextResponse.json({ error: "Forbidden" }, { status: 403 }) as never,
+    );
+
+    const res = await handleResumeChatStream(request(), CHAT_ID);
+
+    expect(res.status).toBe(403);
+    expect(getRun).not.toHaveBeenCalled();
+  });
+
+  // A transient workflow-API failure must not be reported as "nothing to
+  // resume" — that would tell a client with a live run to stop reconnecting.
+  it("returns 502 rather than 204 when the run status lookup throws", async () => {
+    withChat(RUN_ID);
+    vi.mocked(getRun).mockReturnValue({
+      get status() {
+        return Promise.reject(new Error("workflow api down"));
+      },
+      getReadable: vi.fn(),
+    } as never);
+
+    const res = await handleResumeChatStream(request(), CHAT_ID);
+
+    expect(res.status).toBe(502);
+    expect(compareAndSetChatActiveStreamId).not.toHaveBeenCalled();
+  });
+
+  // Upstream open-agents returns this so the client knows which startIndex to
+  // send on its next reconnect; the SDK's WorkflowChatTransport reads it to
+  // compute absolute chunk positions. Without it a reconnect replays from 0.
+  it("advertises the stream tail index so the client can resume precisely", async () => {
+    withChat(RUN_ID);
+    withRun("running");
+
+    const res = await handleResumeChatStream(request(), CHAT_ID);
+
+    expect(res.headers.get("x-workflow-stream-tail-index")).toBe("41");
+  });
+
+  it("still streams when the tail index cannot be read", async () => {
+    withChat(RUN_ID);
+    withRun(
+      "running",
+      vi.fn(() =>
+        Object.assign(new ReadableStream(), {
+          getTailIndex: async () => {
+            throw new Error("unsupported");
+          },
+        }),
+      ),
+    );
+
+    const res = await handleResumeChatStream(request(), CHAT_ID);
+
+    expect(res.status).toBe(200);
+    expect(res.headers.get("x-workflow-stream-tail-index")).toBeNull();
+  });
+});

lib/chat/__tests__/parseStreamStartIndex.test.ts

@@ -0,0 +1,43 @@
+import { describe, it, expect } from "vitest";
+import { NextResponse } from "next/server";
+import { parseStreamStartIndex } from "@/lib/chat/parseStreamStartIndex";
+
+const url = (qs: string) => new URL(`https://api.test/api/chat/abc/stream${qs}`);
+
+describe("parseStreamStartIndex", () => {
+  it("returns undefined when startIndex is absent — a fresh reader gets the whole turn", () => {
+    expect(parseStreamStartIndex(url(""))).toBeUndefined();
+  });
+
+  it("parses a valid non-negative integer", () => {
+    expect(parseStreamStartIndex(url("?startIndex=0"))).toBe(0);
+    expect(parseStreamStartIndex(url("?startIndex=42"))).toBe(42);
+  });
+
+  it("returns a 400 response when startIndex is not a number", () => {
+    const result = parseStreamStartIndex(url("?startIndex=abc"));
+    expect(result).toBeInstanceOf(NextResponse);
+    expect((result as NextResponse).status).toBe(400);
+  });
+
+  // The documented schema is `integer, minimum 0`. A negative value is
+  // meaningful to the underlying SDK (it counts back from the end of a live
+  // stream) but resolves differently on every call, so the contract excludes it.
+  it("returns a 400 response when startIndex is negative", () => {
+    const result = parseStreamStartIndex(url("?startIndex=-5"));
+    expect(result).toBeInstanceOf(NextResponse);
+    expect((result as NextResponse).status).toBe(400);
+  });
+
+  it("returns a 400 response when startIndex is fractional", () => {
+    const result = parseStreamStartIndex(url("?startIndex=1.5"));
+    expect(result).toBeInstanceOf(NextResponse);
+    expect((result as NextResponse).status).toBe(400);
+  });
+
+  it("returns a 400 response when startIndex is present but empty", () => {
+    const result = parseStreamStartIndex(url("?startIndex="));
+    expect(result).toBeInstanceOf(NextResponse);
+    expect((result as NextResponse).status).toBe(400);
+  });
+});

lib/chat/__tests__/validateChatOwnership.test.ts

@@ -0,0 +1,79 @@
+import { describe, it, expect, vi, beforeEach } from "vitest";
+import { NextRequest, NextResponse } from "next/server";
+import { validateChatOwnership } from "@/lib/chat/validateChatOwnership";
+import { validateAuthContext } from "@/lib/auth/validateAuthContext";
+import { selectChats } from "@/lib/supabase/chats/selectChats";
+import { selectSessions } from "@/lib/supabase/sessions/selectSessions";
+
+vi.mock("@/lib/auth/validateAuthContext", () => ({ validateAuthContext: vi.fn() }));
+vi.mock("@/lib/supabase/chats/selectChats", () => ({ selectChats: vi.fn() }));
+vi.mock("@/lib/supabase/sessions/selectSessions", () => ({ selectSessions: vi.fn() }));
+
+const CHAT_ID = "11111111-2222-4333-8444-555555555555";
+const OWNER = "owner-account";
+const ADMIN_TARGET = "22222222-3333-4444-8555-666666666666";
+
+const req = (qs = "") =>
+  new NextRequest(`https://api.test/api/chat/${CHAT_ID}/stream${qs}`, { method: "GET" });
+
+beforeEach(() => {
+  vi.clearAllMocks();
+  vi.mocked(validateAuthContext).mockResolvedValue({ accountId: OWNER, orgId: null } as never);
+  vi.mocked(selectChats).mockResolvedValue([{ id: CHAT_ID, session_id: "sess-1" }] as never);
+  vi.mocked(selectSessions).mockResolvedValue([{ id: "sess-1", account_id: OWNER }] as never);
+});
+
+describe("validateChatOwnership", () => {
+  it("resolves for the owning account", async () => {
+    const result = await validateChatOwnership(req(), CHAT_ID);
+    expect(result).not.toBeInstanceOf(NextResponse);
+  });
+
+  // Without this an org/admin key cannot reach a member's chat — the same gap
+  // DELETE /api/tasks has (chat#1918). validateAuthContext is what decides
+  // whether the caller may actually use the override.
+  it("forwards an account_id query override to validateAuthContext", async () => {
+    await validateChatOwnership(req(`?account_id=${ADMIN_TARGET}`), CHAT_ID);
+
+    expect(validateAuthContext).toHaveBeenCalledWith(
+      expect.anything(),
+      expect.objectContaining({ accountId: ADMIN_TARGET }),
+    );
+  });
+
+  it("lets an approved override through to a chat the key does not personally own", async () => {
+    // validateAuthContext approved the override, so the effective account is
+    // the target — which owns the session.
+    vi.mocked(validateAuthContext).mockResolvedValue({
+      accountId: ADMIN_TARGET,
+      orgId: "org-1",
+    } as never);
+    vi.mocked(selectSessions).mockResolvedValue([
+      { id: "sess-1", account_id: ADMIN_TARGET },
+    ] as never);
+
+    const result = await validateChatOwnership(req(`?account_id=${ADMIN_TARGET}`), CHAT_ID);
+
+    expect(result).not.toBeInstanceOf(NextResponse);
+  });
+
+  it("still 403s when the resolved account does not own the session", async () => {
+    vi.mocked(selectSessions).mockResolvedValue([
+      { id: "sess-1", account_id: "someone-else" },
+    ] as never);
+
+    const result = await validateChatOwnership(req(), CHAT_ID);
+
+    expect(result).toBeInstanceOf(NextResponse);
+    expect((result as NextResponse).status).toBe(403);
+  });
+
+  it("omits the override key entirely when no account_id is supplied", async () => {
+    await validateChatOwnership(req(), CHAT_ID);
+
+    expect(validateAuthContext).toHaveBeenCalledWith(
+      expect.anything(),
+      expect.objectContaining({ accountId: undefined }),
+    );
+  });
+});