From cdfd80d9ce3ea4203fc0446365d9e36f724873f6 Mon Sep 17 00:00:00 2001 From: Sweets Sweetman Date: Mon, 3 Aug 2026 08:29:55 -0500 Subject: [PATCH 1/3] feat(chat): add GET /api/chat/{chatId}/stream to resume an in-progress response MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The endpoint has been documented since the workflow cutover (api-reference/chat/workflow-stream.mdx) but was never implemented — app/api/chat/[chatId]/ contained only stop/. Documented-but-missing drift. It matters now because a long turn's SSE stream can end before the run does. Reproduced on prod 2026-08-02: the stream closed at ~123s with a clean [DONE] and no finish chunk while the workflow ran on to completion, so the client rendered 6 of 13 iterations and froze. The only recovery path was maybeResumeChatStream, which runs inside POST /api/chat — hence a refresh worked and sitting still did not. - app/api/chat/[chatId]/stream/route.ts — GET + OPTIONS, maxDuration 800 to match POST /api/chat, since a resumed stream lives as long as the turn it follows. - lib/chat/handleResumeChatStream.ts — 200 SSE + x-workflow-run-id when the run is live, 204 when there is nothing to resume (clearing a stale active_stream_id on the way), 502 when the status lookup throws. - lib/chat/parseStreamStartIndex.ts — the documented `integer, minimum 0` contract. Negative values are rejected even though the SDK accepts them: it reads those relative to the end of a live stream, which resolves to a different absolute position per call and cannot give a gap-free resume. - lib/chat/validateChatOwnership.ts — extracted from validateStopChatWorkflowRequest so both /stop and /stream enforce the same auth, chat-id and ownership rules from one place. The stop validator is now a thin alias; its behaviour is unchanged. A failed status read returns 502, not 204. Reporting "nothing to resume" for a transient workflow-api blip would tell a client with a live run to stop reconnecting — the exact silent truncation this route exists to prevent. Mirrors reconcileExistingActiveStream's conflict-over-clear stance. Implements docs#286 (adds the startIndex param + 400 to the published contract). Merge order: docs#286 → this → chat client reconnect. Refs recoupable/chat#1923 Co-Authored-By: Claude Opus 5 (1M context) --- app/api/chat/[chatId]/stream/route.ts | 43 ++++++ .../__tests__/handleResumeChatStream.test.ts | 132 ++++++++++++++++++ .../__tests__/parseStreamStartIndex.test.ts | 43 ++++++ lib/chat/handleResumeChatStream.ts | 76 ++++++++++ lib/chat/parseStreamStartIndex.ts | 32 +++++ lib/chat/validateChatOwnership.ts | 55 ++++++++ lib/chat/validateStopChatWorkflowRequest.ts | 53 +++---- 7 files changed, 399 insertions(+), 35 deletions(-) create mode 100644 app/api/chat/[chatId]/stream/route.ts create mode 100644 lib/chat/__tests__/handleResumeChatStream.test.ts create mode 100644 lib/chat/__tests__/parseStreamStartIndex.test.ts create mode 100644 lib/chat/handleResumeChatStream.ts create mode 100644 lib/chat/parseStreamStartIndex.ts create mode 100644 lib/chat/validateChatOwnership.ts diff --git a/app/api/chat/[chatId]/stream/route.ts b/app/api/chat/[chatId]/stream/route.ts new file mode 100644 index 000000000..289f56bdc --- /dev/null +++ b/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 { + const { chatId } = await options.params; + return handleResumeChatStream(request, chatId); +} diff --git a/lib/chat/__tests__/handleResumeChatStream.test.ts b/lib/chat/__tests__/handleResumeChatStream.test.ts new file mode 100644 index 000000000..b666baae2 --- /dev/null +++ b/lib/chat/__tests__/handleResumeChatStream.test.ts @@ -0,0 +1,132 @@ +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(() => new ReadableStream())) { + 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(); + }); +}); diff --git a/lib/chat/__tests__/parseStreamStartIndex.test.ts b/lib/chat/__tests__/parseStreamStartIndex.test.ts new file mode 100644 index 000000000..208d6ec2b --- /dev/null +++ b/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); + }); +}); diff --git a/lib/chat/handleResumeChatStream.ts b/lib/chat/handleResumeChatStream.ts new file mode 100644 index 000000000..7851a6244 --- /dev/null +++ b/lib/chat/handleResumeChatStream.ts @@ -0,0 +1,76 @@ +import { NextRequest, NextResponse } from "next/server"; +import { createUIMessageStreamResponse, type UIMessageChunk } from "ai"; +import { getRun } from "workflow/api"; +import { validateChatOwnership } from "@/lib/chat/validateChatOwnership"; +import { parseStreamStartIndex } from "@/lib/chat/parseStreamStartIndex"; +import { compareAndSetChatActiveStreamId } from "@/lib/chat/compareAndSetChatActiveStreamId"; +import { wrapWorkflowStreamWatcher } from "@/lib/chat/wrapWorkflowStreamWatcher"; +import { errorResponse } from "@/lib/networking/errorResponse"; +import { getCorsHeaders } from "@/lib/networking/getCorsHeaders"; + +const TERMINAL_RUN_STATUSES: ReadonlySet = new Set(["completed", "cancelled", "failed"]); + +/** + * Handles `GET /api/chat/{chatId}/stream` — reconnect to an in-progress + * chat response. + * + * A long turn's SSE stream can end before the run does, which leaves the + * client rendering a half-finished message with no way back in: `POST + * /api/chat` resumes only as a side effect of starting a turn, so today + * recovery needs a full page load (chat#1923). This is the resume path, + * and `startIndex` is what makes it gap-free. + * + * Contract (docs#286): 200 SSE + `x-workflow-run-id`, 204 when there is + * nothing to resume, 400 malformed `startIndex`, 401/403/404 from auth + * and ownership. + * + * @param request - The incoming request. + * @param chatId - Chat id from the route params. + * @returns The resumed stream, 204, or an error response. + */ +export async function handleResumeChatStream( + request: NextRequest, + chatId: string, +): Promise { + const validated = await validateChatOwnership(request, chatId); + if (validated instanceof NextResponse) return validated; + + const startIndex = parseStreamStartIndex(new URL(request.url)); + if (startIndex instanceof NextResponse) return startIndex; + + const activeStreamId = validated.chat.active_stream_id; + if (!activeStreamId) return new NextResponse(null, { status: 204, headers: getCorsHeaders() }); + + const run = getRun(activeStreamId); + + // A failed status read must not be reported as "nothing to resume" — that + // tells a client with a live run to stop reconnecting, which is the exact + // silent-truncation this endpoint exists to prevent. Surface it instead so + // the client retries. Mirrors reconcileExistingActiveStream, which treats a + // transient workflow-api failure as conflict rather than clearing the slot. + let status: string; + try { + status = await run.status; + } catch (error) { + console.error("[handleResumeChatStream] run status lookup failed:", error); + return errorResponse("Failed to read the workflow run", 502); + } + + if (TERMINAL_RUN_STATUSES.has(status)) { + // The run is done, so the slot is stale bookkeeping. Best-effort clear: + // a failed CAS just leaves it for the next request to heal. + const cleared = await compareAndSetChatActiveStreamId(chatId, activeStreamId, null); + if ("error" in cleared) { + console.error("[handleResumeChatStream] failed to clear stale active_stream_id:", cleared); + } + return new NextResponse(null, { status: 204, headers: getCorsHeaders() }); + } + + return createUIMessageStreamResponse({ + stream: wrapWorkflowStreamWatcher( + activeStreamId, + run.getReadable({ startIndex }), + ), + headers: { ...getCorsHeaders(), "x-workflow-run-id": activeStreamId }, + }); +} diff --git a/lib/chat/parseStreamStartIndex.ts b/lib/chat/parseStreamStartIndex.ts new file mode 100644 index 000000000..efb55ae5f --- /dev/null +++ b/lib/chat/parseStreamStartIndex.ts @@ -0,0 +1,32 @@ +import { NextResponse } from "next/server"; +import { validationErrorResponse } from "@/lib/zod/validationErrorResponse"; + +/** + * Parse the optional `startIndex` query param of + * `GET /api/chat/{chatId}/stream`. + * + * Documented as `integer, minimum 0` — the zero-based index of the first + * chunk to return, so a reconnecting client resumes where it left off + * instead of replaying the turn. Absent means "from the beginning", which + * is what a fresh reader (page load, or first watch of a headless run) + * wants. + * + * Negative values are rejected even though the SDK accepts them: it reads + * them relative to the end of a live stream, which resolves to a different + * absolute position on every call and so cannot give a client an exact, + * gap-free resume. + * + * @param url - The request URL carrying the query string. + * @returns The parsed index, `undefined` when absent, or a 400 response. + */ +export function parseStreamStartIndex(url: URL): number | undefined | NextResponse { + const raw = url.searchParams.get("startIndex"); + if (raw === null) return undefined; + + const parsed = Number(raw); + if (raw.trim() === "" || !Number.isInteger(parsed) || parsed < 0) { + return validationErrorResponse("startIndex must be a non-negative integer", ["startIndex"]); + } + + return parsed; +} diff --git a/lib/chat/validateChatOwnership.ts b/lib/chat/validateChatOwnership.ts new file mode 100644 index 000000000..3eb49bfa2 --- /dev/null +++ b/lib/chat/validateChatOwnership.ts @@ -0,0 +1,55 @@ +import { NextRequest, NextResponse } from "next/server"; +import { z } from "zod"; +import { validateAuthContext } from "@/lib/auth/validateAuthContext"; +import type { AuthContext } from "@/lib/auth/validateAuthContext"; +import { selectChats } from "@/lib/supabase/chats/selectChats"; +import { selectSessions } from "@/lib/supabase/sessions/selectSessions"; +import { errorResponse } from "@/lib/networking/errorResponse"; +import { validationErrorResponse } from "@/lib/zod/validationErrorResponse"; +import type { Tables } from "@/types/database.types"; + +export interface ValidatedChatOwnership { + auth: AuthContext; + chat: Tables<"chats">; +} + +const chatIdSchema = z.string().uuid("chatId must be a valid UUID"); + +/** + * Authenticate the caller and confirm they own the chat behind a + * `/api/chat/{chatId}/…` route. + * + * Shared by `POST /api/chat/{chatId}/stop` and + * `GET /api/chat/{chatId}/stream` so both enforce identical auth, + * chat-id validation and ownership semantics. + * + * @param request - The incoming request, carrying the credentials. + * @param chatId - Chat id from the route params. + * @returns The auth context + chat row, or an error response + * (400 malformed id, 401 unauthenticated, 403 not owned, 404 missing). + */ +export async function validateChatOwnership( + request: NextRequest, + chatId: string, +): Promise { + const auth = await validateAuthContext(request); + if (auth instanceof NextResponse) return auth; + + const parsed = chatIdSchema.safeParse(chatId); + if (!parsed.success) { + const firstError = parsed.error.issues[0]; + return validationErrorResponse(firstError.message, firstError.path); + } + + const chats = await selectChats({ id: parsed.data }); + const chat = chats[0]; + if (!chat) return errorResponse("Chat not found", 404); + + const sessions = await selectSessions({ id: chat.session_id }); + if (sessions === null) return errorResponse("Internal server error", 500); + const session = sessions[0]; + if (!session) return errorResponse("Chat not found", 404); + if (session.account_id !== auth.accountId) return errorResponse("Forbidden", 403); + + return { auth, chat }; +} diff --git a/lib/chat/validateStopChatWorkflowRequest.ts b/lib/chat/validateStopChatWorkflowRequest.ts index 75bffe528..dc3d46a9d 100644 --- a/lib/chat/validateStopChatWorkflowRequest.ts +++ b/lib/chat/validateStopChatWorkflowRequest.ts @@ -1,43 +1,26 @@ import { NextRequest, NextResponse } from "next/server"; -import { z } from "zod"; -import { validateAuthContext } from "@/lib/auth/validateAuthContext"; -import type { AuthContext } from "@/lib/auth/validateAuthContext"; -import { selectChats } from "@/lib/supabase/chats/selectChats"; -import { selectSessions } from "@/lib/supabase/sessions/selectSessions"; -import { errorResponse } from "@/lib/networking/errorResponse"; -import { validationErrorResponse } from "@/lib/zod/validationErrorResponse"; -import type { Tables } from "@/types/database.types"; +import { + validateChatOwnership, + type ValidatedChatOwnership, +} from "@/lib/chat/validateChatOwnership"; -export interface ValidatedStopChatWorkflowRequest { - auth: AuthContext; - chat: Tables<"chats">; -} - -const chatIdSchema = z.string().uuid("chatId must be a valid UUID"); +export type ValidatedStopChatWorkflowRequest = ValidatedChatOwnership; -/** Validates POST /api/chat/{chatId}/stop: auth, chatId format, and chat + session-ownership lookup. */ +/** + * Validates POST /api/chat/{chatId}/stop: auth, chatId format, and chat + + * session-ownership lookup. + * + * Thin alias over `validateChatOwnership`, which `GET + * /api/chat/{chatId}/stream` shares — both routes must agree on who may + * touch a chat, so the rule lives in one place. + * + * @param request - The incoming request. + * @param chatId - Chat id from the route params. + * @returns The auth context + chat row, or an error response. + */ export async function validateStopChatWorkflowRequest( request: NextRequest, chatId: string, ): Promise { - const auth = await validateAuthContext(request); - if (auth instanceof NextResponse) return auth; - - const parsed = chatIdSchema.safeParse(chatId); - if (!parsed.success) { - const firstError = parsed.error.issues[0]; - return validationErrorResponse(firstError.message, firstError.path); - } - - const chats = await selectChats({ id: parsed.data }); - const chat = chats[0]; - if (!chat) return errorResponse("Chat not found", 404); - - const sessions = await selectSessions({ id: chat.session_id }); - if (sessions === null) return errorResponse("Internal server error", 500); - const session = sessions[0]; - if (!session) return errorResponse("Chat not found", 404); - if (session.account_id !== auth.accountId) return errorResponse("Forbidden", 403); - - return { auth, chat }; + return validateChatOwnership(request, chatId); } From bcd819bb4d070804ef14fcf4b18a0e85de67f512 Mon Sep 17 00:00:00 2001 From: Sweets Sweetman Date: Mon, 3 Aug 2026 09:28:39 -0500 Subject: [PATCH 2/3] fix(chat): claim active_stream_id on headless runs so they are resumable MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Preview testing of the resume route turned up that GET /api/chat/{chatId}/stream returns 204 for a live headless run: lib/chat/runs/ never sets chats.active_stream_id, and the route keys on it. That contradicts the published contract, which cross-references the two in both directions — POST /api/chat/runs says "read the result via GET /api/chat/{chatId}/stream (resume the stream)", and the stream endpoint says "start a headless run, then pass the returned chatId here to watch its output live". handleStartChatRun's own comment says the same. The intent was always there; only the slot claim was missing. Claims the slot right after start(). The chat is freshly provisioned so nothing contends for it, and the workflow's clearChatActiveStream already releases it on run end — so this just closes the loop symmetrically with the interactive path. Best-effort: a failed claim costs resumability, not the run. Refs recoupable/chat#1923 Co-Authored-By: Claude Opus 5 (1M context) --- .../runs/__tests__/handleStartChatRun.test.ts | 14 ++++++++++++++ lib/chat/runs/handleStartChatRun.ts | 16 ++++++++++++++++ 2 files changed, 30 insertions(+) diff --git a/lib/chat/runs/__tests__/handleStartChatRun.test.ts b/lib/chat/runs/__tests__/handleStartChatRun.test.ts index 3f1c8555b..999b12131 100644 --- a/lib/chat/runs/__tests__/handleStartChatRun.test.ts +++ b/lib/chat/runs/__tests__/handleStartChatRun.test.ts @@ -8,6 +8,7 @@ import { mintEphemeralAccountKey } from "@/lib/keys/mintEphemeralAccountKey"; import { deleteApiKey } from "@/lib/supabase/account_api_keys/deleteApiKey"; import { buildRunAgentInput } from "@/lib/chat/buildRunAgentInput"; import { start } from "workflow/api"; +import { compareAndSetChatActiveStreamId } from "@/lib/chat/compareAndSetChatActiveStreamId"; vi.mock("@/lib/networking/getCorsHeaders", () => ({ getCorsHeaders: vi.fn(() => ({ "Access-Control-Allow-Origin": "*" })), @@ -31,6 +32,9 @@ vi.mock("workflow/api", () => ({ start: vi.fn(), })); vi.mock("@/app/lib/workflows/runAgentWorkflow", () => ({ runAgentWorkflow: vi.fn() })); +vi.mock("@/lib/chat/compareAndSetChatActiveStreamId", () => ({ + compareAndSetChatActiveStreamId: vi.fn(async () => ({ ok: true, claimed: true })), +})); const req = () => new NextRequest("https://x.test/api/chat/generate", { @@ -121,4 +125,14 @@ describe("handleStartChatRun", () => { expect(mintEphemeralAccountKey).not.toHaveBeenCalled(); expect(deleteApiKey).not.toHaveBeenCalled(); }); + + // The published contract cross-references this: "start a headless run, then + // pass the returned chatId to GET /api/chat/{chatId}/stream to watch its + // output live". That route keys on chats.active_stream_id, so a headless run + // that never claims the slot is unresumable (chat#1923). + it("claims chats.active_stream_id with the run id so the run is resumable", async () => { + await handleStartChatRun({} as never); + + expect(compareAndSetChatActiveStreamId).toHaveBeenCalledWith("chat-1", null, "wrun_abc"); + }); }); diff --git a/lib/chat/runs/handleStartChatRun.ts b/lib/chat/runs/handleStartChatRun.ts index 41b68d1c7..4122834c8 100644 --- a/lib/chat/runs/handleStartChatRun.ts +++ b/lib/chat/runs/handleStartChatRun.ts @@ -8,6 +8,7 @@ import { mintEphemeralAccountKey } from "@/lib/keys/mintEphemeralAccountKey"; import { deleteApiKey } from "@/lib/supabase/account_api_keys/deleteApiKey"; import { buildRunAgentInput } from "@/lib/chat/buildRunAgentInput"; import { runAgentWorkflow } from "@/app/lib/workflows/runAgentWorkflow"; +import { compareAndSetChatActiveStreamId } from "@/lib/chat/compareAndSetChatActiveStreamId"; /** Default title for the session a headless run provisions (no caller-supplied title). */ const DEFAULT_RUN_SESSION_TITLE = "Scheduled generation"; @@ -69,6 +70,21 @@ export async function handleStartChatRun(request: NextRequest): Promise Date: Mon, 3 Aug 2026 10:00:25 -0500 Subject: [PATCH 3/3] feat(chat): admin account_id override + stream tail index on the resume route MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two gaps found reviewing this route against upstream open-agents. 1. Admin override. validateChatOwnership called validateAuthContext with no override options, so an org/admin key got a 403 on a chat it legitimately administers — the same defect as DELETE /api/tasks (chat#1918). Now reads `account_id` from the query string and passes it through. Query rather than body: both /stream (GET) and /stop (POST) carry their id in the path and parse no body, so a query param is the one channel that works for both without consuming the request. validateAuthContext still decides whether the caller may use the override, so this does not weaken the check — it just stops discarding a legitimate one. Because the validator is shared, this fixes POST /api/chat/{chatId}/stop at the same time, which had the identical limitation before this PR. 2. x-workflow-stream-tail-index. Upstream returns readable.getTailIndex() so a client knows which startIndex to send on its next reconnect; the SDK's WorkflowChatTransport reads the same header to compute absolute chunk positions. Without it a reconnect replays from chunk zero. getTailIndex() is available on the WorkflowReadableStream in workflow@4.2.4. Best-effort: if the runtime cannot report a tail index we still stream. A replaying client beats no client. Deliberately unchanged, having compared both against upstream: - A failed getRun still returns 502 and keeps the slot. Upstream clears the slot and returns 204 on any error; that would tell a client with a live run to stop reconnecting, which is the silent truncation this route exists to prevent. Ours mirrors reconcileExistingActiveStream. - wrapWorkflowStreamWatcher stays instead of upstream's createCancelableReadableStream: ours also reconciles orphaned tool-calls and propagates cancel to the run. Full api suite 4,329 pass. Refs recoupable/chat#1923 Co-Authored-By: Claude Opus 5 (1M context) --- .../__tests__/handleResumeChatStream.test.ts | 36 ++++++++- .../__tests__/validateChatOwnership.test.ts | 79 +++++++++++++++++++ lib/chat/handleResumeChatStream.ts | 25 ++++-- lib/chat/validateChatOwnership.ts | 12 ++- 4 files changed, 145 insertions(+), 7 deletions(-) create mode 100644 lib/chat/__tests__/validateChatOwnership.test.ts diff --git a/lib/chat/__tests__/handleResumeChatStream.test.ts b/lib/chat/__tests__/handleResumeChatStream.test.ts index b666baae2..21a984ccb 100644 --- a/lib/chat/__tests__/handleResumeChatStream.test.ts +++ b/lib/chat/__tests__/handleResumeChatStream.test.ts @@ -25,7 +25,10 @@ function withChat(activeStreamId: string | null) { } as never); } -function withRun(status: string, getReadable = vi.fn(() => new ReadableStream())) { +function withRun( + status: string, + getReadable = vi.fn(() => Object.assign(new ReadableStream(), { getTailIndex: async () => 41 })), +) { vi.mocked(getRun).mockReturnValue({ get status() { return Promise.resolve(status); @@ -129,4 +132,35 @@ describe("handleResumeChatStream", () => { 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(); + }); }); diff --git a/lib/chat/__tests__/validateChatOwnership.test.ts b/lib/chat/__tests__/validateChatOwnership.test.ts new file mode 100644 index 000000000..152d6620d --- /dev/null +++ b/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 }), + ); + }); +}); diff --git a/lib/chat/handleResumeChatStream.ts b/lib/chat/handleResumeChatStream.ts index 7851a6244..31edc942d 100644 --- a/lib/chat/handleResumeChatStream.ts +++ b/lib/chat/handleResumeChatStream.ts @@ -66,11 +66,26 @@ export async function handleResumeChatStream( return new NextResponse(null, { status: 204, headers: getCorsHeaders() }); } + const readable = run.getReadable({ startIndex }); + + // Tell the client where this read ends so its next reconnect can resume + // exactly there instead of replaying from chunk zero. Upstream open-agents + // returns the same header, and the SDK's WorkflowChatTransport reads it to + // compute absolute chunk positions. Best-effort: if the runtime can't report + // a tail index we still stream — a replaying client beats no client. + let tailIndex: number | undefined; + try { + tailIndex = await readable.getTailIndex(); + } catch (error) { + console.error("[handleResumeChatStream] getTailIndex failed:", error); + } + return createUIMessageStreamResponse({ - stream: wrapWorkflowStreamWatcher( - activeStreamId, - run.getReadable({ startIndex }), - ), - headers: { ...getCorsHeaders(), "x-workflow-run-id": activeStreamId }, + stream: wrapWorkflowStreamWatcher(activeStreamId, readable), + headers: { + ...getCorsHeaders(), + "x-workflow-run-id": activeStreamId, + ...(tailIndex === undefined ? {} : { "x-workflow-stream-tail-index": String(tailIndex) }), + }, }); } diff --git a/lib/chat/validateChatOwnership.ts b/lib/chat/validateChatOwnership.ts index 3eb49bfa2..0aef35e8c 100644 --- a/lib/chat/validateChatOwnership.ts +++ b/lib/chat/validateChatOwnership.ts @@ -23,6 +23,15 @@ const chatIdSchema = z.string().uuid("chatId must be a valid UUID"); * `GET /api/chat/{chatId}/stream` so both enforce identical auth, * chat-id validation and ownership semantics. * + * Honours the `account_id` **query** override, so an org/admin key can act + * on a member account's chat. Both routes carry their id in the path and + * parse no body, so a query param is the channel that works for `GET` and + * `POST` alike without consuming the request body. `validateAuthContext` is + * what decides whether the caller may actually use the override — passing it + * through does not weaken the check, and omitting it was why an admin key got + * a 403 on a chat it legitimately administers (same gap as `DELETE + * /api/tasks`, chat#1918). + * * @param request - The incoming request, carrying the credentials. * @param chatId - Chat id from the route params. * @returns The auth context + chat row, or an error response @@ -32,7 +41,8 @@ export async function validateChatOwnership( request: NextRequest, chatId: string, ): Promise { - const auth = await validateAuthContext(request); + const accountIdOverride = new URL(request.url).searchParams.get("account_id") ?? undefined; + const auth = await validateAuthContext(request, { accountId: accountIdOverride }); if (auth instanceof NextResponse) return auth; const parsed = chatIdSchema.safeParse(chatId);