feat(api): split chat-workflow resume into GET /api/chat/[chatId]/stream

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

@@ -0,0 +1,41 @@
+import type { NextRequest } from "next/server";
+import { NextResponse } from "next/server";
+import { getCorsHeaders } from "@/lib/networking/getCorsHeaders";
+import { handleChatStreamResume } from "@/lib/chat/handleChatStreamResume";
+
+export const maxDuration = 800;
+
+/**
+ * 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
+ *
+ * Resume/reconnect endpoint for the durable chat-workflow agent loop. Returns
+ * the live UIMessage stream when a run is still in flight, or 204 when there
+ * is nothing to resume. This is the only resume path — POST /api/chat/workflow
+ * never resumes.
+ *
+ * Authentication: x-api-key or Authorization bearer (caller must own the chat).
+ *
+ * @param request - The incoming request.
+ * @param context - Next.js route context.
+ * @param context.params - Async route params containing the chat id.
+ * @returns A streaming Response (200), an empty 204, or an error response.
+ */
+export async function GET(
+  request: NextRequest,
+  context: { params: Promise<{ chatId: string }> },
+): Promise<Response> {
+  const { chatId } = await context.params;
+  return handleChatStreamResume(request, chatId);
+}

lib/chat/__tests__/handleChatStreamResume.test.ts

@@ -0,0 +1,80 @@
+import { describe, it, expect, vi, beforeEach } from "vitest";
+import { NextRequest, NextResponse } from "next/server";
+
+import { handleChatStreamResume } from "@/lib/chat/handleChatStreamResume";
+import { validateGetChatStreamRequest } from "@/lib/chat/validateGetChatStreamRequest";
+import { reconcileExistingActiveStream } from "@/lib/chat/reconcileExistingActiveStream";
+
+vi.mock("@/lib/chat/validateGetChatStreamRequest", () => ({
+  validateGetChatStreamRequest: vi.fn(),
+}));
+vi.mock("@/lib/chat/reconcileExistingActiveStream", () => ({
+  reconcileExistingActiveStream: vi.fn(),
+}));
+vi.mock("@/lib/networking/getCorsHeaders", () => ({
+  getCorsHeaders: vi.fn(() => ({ "Access-Control-Allow-Origin": "*" })),
+}));
+
+const ACCOUNT_ID = "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa";
+const CHAT_ID = "11111111-1111-1111-1111-111111111111";
+
+function makeRequest(): NextRequest {
+  return new NextRequest(`http://localhost/api/chat/${CHAT_ID}/stream`, {
+    method: "GET",
+    headers: { "x-api-key": "test-key" },
+  });
+}
+
+function mockValidatedChat(activeStreamId: string | null) {
+  vi.mocked(validateGetChatStreamRequest).mockResolvedValue({
+    chat: { id: CHAT_ID, active_stream_id: activeStreamId } as never,
+    accountId: ACCOUNT_ID,
+  });
+}
+
+beforeEach(() => vi.clearAllMocks());
+
+describe("handleChatStreamResume", () => {
+  it("passes through the validator's error response", async () => {
+    vi.mocked(validateGetChatStreamRequest).mockResolvedValue(
+      NextResponse.json({ status: "error", error: "Forbidden" }, { status: 403 }),
+    );
+    const res = await handleChatStreamResume(makeRequest(), CHAT_ID);
+    expect(res.status).toBe(403);
+    expect(reconcileExistingActiveStream).not.toHaveBeenCalled();
+  });
+
+  it("returns 204 when the chat has no active stream id", async () => {
+    mockValidatedChat(null);
+    const res = await handleChatStreamResume(makeRequest(), CHAT_ID);
+    expect(res.status).toBe(204);
+    expect(reconcileExistingActiveStream).not.toHaveBeenCalled();
+  });
+
+  it("returns 200 with the live stream and x-workflow-run-id when resumable", async () => {
+    mockValidatedChat("wrun_live");
+    vi.mocked(reconcileExistingActiveStream).mockResolvedValue({
+      action: "resume",
+      runId: "wrun_live",
+      stream: new ReadableStream(),
+    });
+    const res = await handleChatStreamResume(makeRequest(), CHAT_ID);
+    expect(res.status).toBe(200);
+    expect(res.headers.get("content-type") ?? "").toMatch(/text\/event-stream/);
+    expect(res.headers.get("x-workflow-run-id")).toBe("wrun_live");
+  });
+
+  it("returns 204 when the run is terminal (reconcile=ready, stale id cleared)", async () => {
+    mockValidatedChat("wrun_done");
+    vi.mocked(reconcileExistingActiveStream).mockResolvedValue({ action: "ready" });
+    const res = await handleChatStreamResume(makeRequest(), CHAT_ID);
+    expect(res.status).toBe(204);
+  });
+
+  it("returns 204 when the probe is indeterminate (reconcile=conflict)", async () => {
+    mockValidatedChat("wrun_unknown");
+    vi.mocked(reconcileExistingActiveStream).mockResolvedValue({ action: "conflict" });
+    const res = await handleChatStreamResume(makeRequest(), CHAT_ID);
+    expect(res.status).toBe(204);
+  });
+});

lib/chat/__tests__/handleChatWorkflowStream.test.ts

@@ -8,7 +8,7 @@ import { selectChats } from "@/lib/supabase/chats/selectChats";
 import { isSandboxActive } from "@/lib/sandbox/isSandboxActive";
 import { updateSession } from "@/lib/supabase/sessions/updateSession";
 import { compareAndSetChatActiveStreamId } from "@/lib/chat/compareAndSetChatActiveStreamId";
-import { maybeResumeChatStream } from "@/lib/chat/maybeResumeChatStream";
+import { reconcileExistingActiveStream } from "@/lib/chat/reconcileExistingActiveStream";
 import { persistLatestUserMessage } from "@/lib/chat/persistLatestUserMessage";
 import { start, getRun } from "workflow/api";
 
@@ -23,8 +23,8 @@ vi.mock("@/lib/supabase/sessions/updateSession", () => ({ updateSession: vi.fn()
 vi.mock("@/lib/sandbox/buildActiveLifecycleUpdate", () => ({
   buildActiveLifecycleUpdate: vi.fn(() => ({})),
 }));
-vi.mock("@/lib/chat/maybeResumeChatStream", () => ({
-  maybeResumeChatStream: vi.fn(),
+vi.mock("@/lib/chat/reconcileExistingActiveStream", () => ({
+  reconcileExistingActiveStream: vi.fn(),
 }));
 vi.mock("@/lib/chat/persistLatestUserMessage", () => ({
   persistLatestUserMessage: vi.fn(),
@@ -109,8 +109,9 @@ function mockStartedRun(runId = "wrun_test_run_1") {
 
 beforeEach(() => {
   vi.clearAllMocks();
-  // Default: maybeResumeChatStream returns null (no resume / no active stream)
-  vi.mocked(maybeResumeChatStream).mockResolvedValue(null);
+  // Default: a stale/terminal active stream that self-heals to "ready" so the
+  // POST proceeds. Only invoked when the chat has an active_stream_id.
+  vi.mocked(reconcileExistingActiveStream).mockResolvedValue({ action: "ready" });
 });
 
 describe("handleChatWorkflowStream", () => {
@@ -166,31 +167,41 @@ describe("handleChatWorkflowStream", () => {
     });
   });
 
-  describe("resume / conflict via maybeResumeChatStream", () => {
+  describe("active stream handling (start-only; never resumes)", () => {
     beforeEach(() => {
       mockValidated();
       mockSessionOwnedActive();
       mockChatOwned({ active_stream_id: "wrun_existing" });
     });
 
-    it("returns the resume response when maybeResumeChatStream yields one", async () => {
-      const resumeResponse = new Response("ok", {
-        status: 200,
-        headers: { "x-workflow-run-id": "wrun_existing" },
+    it("returns 409 without starting when a run is live (reconcile=resume)", async () => {
+      vi.mocked(reconcileExistingActiveStream).mockResolvedValue({
+        action: "resume",
+        runId: "wrun_existing",
+        stream: new ReadableStream(),
       });
-      vi.mocked(maybeResumeChatStream).mockResolvedValue(resumeResponse);
       const res = await handleChatWorkflowStream(makeRequest());
-      expect(res.headers.get("x-workflow-run-id")).toBe("wrun_existing");
+      expect(res.status).toBe(409);
       expect(start).not.toHaveBeenCalled();
     });
 
-    it("returns the conflict response when maybeResumeChatStream yields 409", async () => {
-      const conflict = NextResponse.json({ status: "error", error: "conflict" }, { status: 409 });
-      vi.mocked(maybeResumeChatStream).mockResolvedValue(conflict);
+    it("returns 503 (retryable) without starting when reconcile is indeterminate (conflict)", async () => {
+      vi.mocked(reconcileExistingActiveStream).mockResolvedValue({ action: "conflict" });
       const res = await handleChatWorkflowStream(makeRequest());
-      expect(res.status).toBe(409);
+      expect(res.status).toBe(503);
       expect(start).not.toHaveBeenCalled();
     });
+
+    it("starts a fresh run when a terminal stale id self-heals (reconcile=ready)", async () => {
+      vi.mocked(reconcileExistingActiveStream).mockResolvedValue({ action: "ready" });
+      vi.mocked(compareAndSetChatActiveStreamId)
+        .mockResolvedValueOnce({ ok: true, claimed: true })
+        .mockResolvedValueOnce({ ok: true, claimed: true });
+      mockStartedRun();
+      const res = await handleChatWorkflowStream(makeRequest());
+      expect(res.status).toBe(200);
+      expect(start).toHaveBeenCalled();
+    });
   });
 
   describe("placeholder CAS before start", () => {

lib/chat/__tests__/validateGetChatStreamRequest.test.ts

@@ -0,0 +1,112 @@
+import { describe, it, expect, vi, beforeEach } from "vitest";
+import { NextRequest, NextResponse } from "next/server";
+
+import { validateGetChatStreamRequest } from "@/lib/chat/validateGetChatStreamRequest";
+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() }));
+vi.mock("@/lib/networking/getCorsHeaders", () => ({
+  getCorsHeaders: vi.fn(() => ({ "Access-Control-Allow-Origin": "*" })),
+}));
+
+const ACCOUNT_ID = "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa";
+const OTHER_ACCOUNT_ID = "bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb";
+const SESSION_ID = "22222222-2222-4222-8222-222222222222";
+const CHAT_ID = "11111111-1111-4111-8111-111111111111";
+
+function makeRequest(): NextRequest {
+  return new NextRequest(`http://localhost/api/chat/${CHAT_ID}/stream`, {
+    method: "GET",
+    headers: { "x-api-key": "test-key" },
+  });
+}
+
+function mockAuthed() {
+  vi.mocked(validateAuthContext).mockResolvedValue({
+    accountId: ACCOUNT_ID,
+    orgId: null,
+    authToken: "test-key",
+  });
+}
+
+beforeEach(() => vi.clearAllMocks());
+
+describe("validateGetChatStreamRequest", () => {
+  it("returns 400 when chatId is empty", async () => {
+    const res = await validateGetChatStreamRequest(makeRequest(), "");
+    expect(res).toBeInstanceOf(NextResponse);
+    expect((res as NextResponse).status).toBe(400);
+  });
+
+  it("returns 400 when chatId is not a valid UUID", async () => {
+    const res = await validateGetChatStreamRequest(makeRequest(), "not-a-uuid");
+    expect((res as NextResponse).status).toBe(400);
+  });
+
+  it("passes through the auth error response", async () => {
+    vi.mocked(validateAuthContext).mockResolvedValue(
+      NextResponse.json({ status: "error", error: "Unauthorized" }, { status: 401 }),
+    );
+    const res = await validateGetChatStreamRequest(makeRequest(), CHAT_ID);
+    expect((res as NextResponse).status).toBe(401);
+  });
+
+  it("returns 404 when the chat does not exist", async () => {
+    mockAuthed();
+    vi.mocked(selectChats).mockResolvedValue([]);
+    const res = await validateGetChatStreamRequest(makeRequest(), CHAT_ID);
+    expect((res as NextResponse).status).toBe(404);
+  });
+
+  it("returns 500 when the chat lookup errors (selectChats null)", async () => {
+    mockAuthed();
+    vi.mocked(selectChats).mockResolvedValue(null);
+    const res = await validateGetChatStreamRequest(makeRequest(), CHAT_ID);
+    expect((res as NextResponse).status).toBe(500);
+  });
+
+  it("returns 500 when the session lookup errors", async () => {
+    mockAuthed();
+    vi.mocked(selectChats).mockResolvedValue([{ id: CHAT_ID, session_id: SESSION_ID } as never]);
+    vi.mocked(selectSessions).mockResolvedValue(null);
+    const res = await validateGetChatStreamRequest(makeRequest(), CHAT_ID);
+    expect((res as NextResponse).status).toBe(500);
+  });
+
+  it("returns 404 when the session does not exist", async () => {
+    mockAuthed();
+    vi.mocked(selectChats).mockResolvedValue([{ id: CHAT_ID, session_id: SESSION_ID } as never]);
+    vi.mocked(selectSessions).mockResolvedValue([]);
+    const res = await validateGetChatStreamRequest(makeRequest(), CHAT_ID);
+    expect((res as NextResponse).status).toBe(404);
+  });
+
+  it("returns 403 when the caller does not own the session", async () => {
+    mockAuthed();
+    vi.mocked(selectChats).mockResolvedValue([{ id: CHAT_ID, session_id: SESSION_ID } as never]);
+    vi.mocked(selectSessions).mockResolvedValue([
+      { id: SESSION_ID, account_id: OTHER_ACCOUNT_ID } as never,
+    ]);
+    const res = await validateGetChatStreamRequest(makeRequest(), CHAT_ID);
+    expect((res as NextResponse).status).toBe(403);
+  });
+
+  it("returns the chat row when the caller owns it", async () => {
+    mockAuthed();
+    vi.mocked(selectChats).mockResolvedValue([
+      { id: CHAT_ID, session_id: SESSION_ID, active_stream_id: "wrun_x" } as never,
+    ]);
+    vi.mocked(selectSessions).mockResolvedValue([
+      { id: SESSION_ID, account_id: ACCOUNT_ID } as never,
+    ]);
+    const res = await validateGetChatStreamRequest(makeRequest(), CHAT_ID);
+    expect(res).not.toBeInstanceOf(NextResponse);
+    if (res instanceof NextResponse) return;
+    expect(res.chat.id).toBe(CHAT_ID);
+    expect(res.accountId).toBe(ACCOUNT_ID);
+  });
+});