ai-chat.integration.test.ts

ai-chat.integration.test.tstests/integration/ai-chat.integration.test.tsraw ← integration21,884 B← Back to the app
1/**
2 * TASK-0122 — Ask AI chat (SSE) integration suite.
3 *
4 * Runs the full stack (AppModule + Postgres + Redis) against the
5 * deterministic `MockProvider` (env `AI_PROVIDER=mock`) and covers the
6 * spec's integration acceptance criteria:
7 *
8 *   - conversation CRUD + ownership isolation (user A cannot GET user B's
9 *     conversation → 404 — existence hidden across the boundary);
10 *   - the SSE stream contract: content-type, frame sequence
11 *     (delta* → usage → done), auto-title, context inclusion;
12 *   - context binding: the mock echoes the system-prompt tail, so a question
13 *     about a page's seeded content answers with THAT content, and a user
14 *     without read access to the referenced page gets a context-unavailable
15 *     notice — NOT the page content (no page text in the prompt);
16 *   - budget exhaustion → canonical 402 AI_BUDGET_EXCEEDED (pre-stream);
17 *   - retry (`retryOf`) re-sends without duplicating the user message row;
18 *   - abort → partial assistant message with `metadata.truncated = true`;
19 *   - per-user 30 msg/min rate limit → 429 RATE_LIMITED + retryAfterSeconds;
20 *   - the `ai.conversation_created` audit event.
21 */
22// Config must be fixed BEFORE the app boots (loadEnv + the workspace cache are
23// both lazily primed on first use).
24process.env.AI_PROVIDER = "mock";
25process.env.SEARCH_VECTOR_DIMENSION = "8";
26// Pace the mock's deltas (one per word) so the mid-stream-cancel test can
27// abort BETWEEN deltas deterministically. 25 ms × ~13 deltas ≈ 325 ms per
28// stream — the client lands its abort (after 3 deltas ≈ 75 ms) well before
29// the server finishes, so the partial-persist path is guaranteed.
30process.env.AI_MOCK_DELTA_DELAY_MS = "25";
31 
32import { afterAll, beforeAll, describe, expect, it } from "vitest";
33import { createTestApp, type TestAppHandle } from "./helpers/api";
34import { addMemberViaSeam, authHeaders, countAuditEvents, createWorkspace, createUser } from "./helpers/seed";
35 
36let app: TestAppHandle;
37 
38beforeAll(async () => {
39  app = await createTestApp();
40}, 180_000);
41afterAll(async () => {
42  await app?.close();
43});
44 
45// ---------------------------------------------------------------------------
46// SSE helpers
47// ---------------------------------------------------------------------------
48 
49type SseEvent = {
50  type: "delta" | "tool_call" | "usage" | "error" | "done";
51  text?: string;
52  tokensIn?: number;
53  tokensOut?: number;
54  code?: string;
55  message?: string;
56  retryable?: boolean;
57  retryAfterSeconds?: number;
58  assistantMessageId?: string;
59  userMessageId?: string;
60  reason?: string;
61};
62 
63/** POST an SSE message, collect the frames; `abortAfter` closes the socket
64 *  (client-cancel simulation) once ≥N delta frames arrived. */
65async function streamMessage(
66  accessToken: string,
67  workspaceId: string,
68  conversationId: string,
69  body: Record<string, unknown>,
70  opts: { abortAfterDeltas?: number } = {},
71): Promise<{ status: number; events: SseEvent[]; contentType: string }> {
72  const controller = new AbortController();
73  let deltas = 0;
74  const res = await fetch(`${app.baseUrl}/api/v1/workspaces/${workspaceId}/ai/conversations/${conversationId}/messages`, {
75    method: "POST",
76    headers: { "Content-Type": "application/json", Authorization: `Bearer ${accessToken}` },
77    body: JSON.stringify(body),
78    signal: controller.signal,
79  });
80  const contentType = res.headers.get("content-type") ?? "";
81  if (!res.body) throw new Error("no SSE body");
82  const events: SseEvent[] = [];
83  const reader = res.body.getReader();
84  const decoder = new TextDecoder();
85  let buf = "";
86  try {
87    for (;;) {
88      const chunk = await reader.read();
89      if (chunk.done) break;
90      buf += decoder.decode(chunk.value, { stream: true });
91      let idx: number;
92      while ((idx = buf.indexOf("\n\n")) !== -1) {
93        const frame = buf.slice(0, idx);
94        buf = buf.slice(idx + 2);
95        for (const line of frame.split("\n")) {
96          if (!line.startsWith("data:")) continue; // skip `: keepalive` comments
97          events.push(JSON.parse(line.slice(5).trim()) as SseEvent);
98        }
99        if (opts.abortAfterDeltas !== undefined) {
100          deltas = events.filter((e) => e.type === "delta").length;
101          if (deltas >= opts.abortAfterDeltas) {
102            controller.abort();
103            break;
104          }
105        }
106      }
107      if (controller.signal.aborted) break;
108    }
109  } catch (err) {
110    // client-side abort: the pending read() rejects with an AbortError —
111    // that IS the cancellation we wanted (the frames seen so far are kept).
112    if (!controller.signal.aborted || !(err instanceof Error && err.name === "AbortError")) throw err;
113  }
114  return { status: res.status, events, contentType };
115}
116 
117async function createConversation(
118  accessToken: string,
119  workspaceId: string,
120  body?: Record<string, unknown>,
121): Promise<{ status: number; id?: string }> {
122  const res = await app.request("POST", `/api/v1/workspaces/${workspaceId}/ai/conversations`, {
123    headers: authHeaders(accessToken),
124    body: body ?? {},
125  });
126  const out = (await res.json().catch(() => ({}))) as { id?: string };
127  return { status: res.status, id: out.id };
128}
129 
130/** Create a page + N PARAGRAPH blocks (distinctive text) via the API. */
131async function createPageWithBlocks(
132  accessToken: string,
133  workspaceId: string,
134  title: string,
135  blockTexts: string[],
136): Promise<{ pageId: string; blockId: string }> {
137  const pageRes = await app.request("POST", `/api/v1/workspaces/${workspaceId}/pages`, {
138    headers: authHeaders(accessToken),
139    body: { title },
140  });
141  const page = (await pageRes.json()) as { id: string };
142  let prevId: string | undefined;
143  for (const text of blockTexts) {
144    const r = await app.request("POST", `/api/v1/pages/${page.id}/blocks`, {
145      headers: authHeaders(accessToken),
146      body: { type: "PARAGRAPH", content: { text }, ...(prevId ? { afterBlockId: prevId } : {}) },
147    });
148    expect(r.status).toBe(201);
149    const b = (await r.json()) as { id: string };
150    prevId = b.id;
151  }
152  const blocks = await app.prisma.block.findMany({ where: { pageId: page.id, deletedAt: null }, orderBy: { position: "asc" } });
153  return { pageId: page.id, blockId: blocks[0]!.id };
154}
155 
156/** Concatenated assistant text from the SSE events. */
157const deltaText = (events: SseEvent[]) =>
158  events.filter((e) => e.type === "delta").map((e) => e.text ?? "").join("");
159 
160// ---------------------------------------------------------------------------
161// Tests
162// ---------------------------------------------------------------------------
163 
164describe("TASK-0122 Ask AI (SSE) — conversation CRUD + ownership", () => {
165  it("create → list → get → rename → delete round-trips (own scope)", async () => {
166    const owner = await createUser(app.request, { name: "AI Chat Owner" });
167    const { workspaceId } = await createWorkspace(app.request, authHeaders(owner.accessToken!), "AI Chat WS");
168    const h = authHeaders(owner.accessToken!);
169 
170    const created = await createConversation(owner.accessToken!, workspaceId, { title: "First" });
171    expect(created.status).toBe(201);
172    expect(created.id).toBeTruthy();
173 
174    // audit event
175    expect(await countAuditEvents(app.prisma, { action: "ai.conversation_created" })).toBeGreaterThanOrEqual(1);
176 
177    // list (most recent 50, own scope)
178    const listRes = await app.request("GET", `/api/v1/workspaces/${workspaceId}/ai/conversations`, { headers: h });
179    expect(listRes.status).toBe(200);
180    const list = (await listRes.json()) as { data: Array<{ id: string; title: string }> };
181    expect(list.data.some((c) => c.id === created.id)).toBe(true);
182 
183    // rename
184    const patchRes = await app.request("PATCH", `/api/v1/ai/conversations/${created.id}`, {
185      headers: h,
186      body: { title: "Renamed" },
187    });
188    expect(patchRes.status).toBe(200);
189 
190    // delete (soft) → 404 thereafter
191    const delRes = await app.request("DELETE", `/api/v1/ai/conversations/${created.id}`, { headers: h });
192    expect(delRes.status).toBe(200);
193    const getRes = await app.request("GET", `/api/v1/ai/conversations/${created.id}`, { headers: h });
194    expect(getRes.status).toBe(404);
195  });
196 
197  it("user A cannot read / write user B's conversation (404 — existence hidden)", async () => {
198    const a = await createUser(app.request, { name: "AI Chat A" });
199    const b = await createUser(app.request, { name: "AI Chat B" });
200    const { workspaceId } = await createWorkspace(app.request, authHeaders(a.accessToken!), "AI Chat WS2");
201    await addMemberViaSeam(app.request, app.internalServiceToken, workspaceId, b.email, "MEMBER");
202 
203    const mine = await createConversation(a.accessToken!, workspaceId, { title: "A's convo" });
204    expect(mine.id).toBeTruthy();
205 
206    // B: GET → 404, DELETE → 404
207    const getRes = await app.request("GET", `/api/v1/ai/conversations/${mine.id}`, {
208      headers: authHeaders(b.accessToken!),
209    });
210    expect(getRes.status).toBe(404);
211    const delRes = await app.request("DELETE", `/api/v1/ai/conversations/${mine.id}`, {
212      headers: authHeaders(b.accessToken!),
213    });
214    expect(delRes.status).toBe(404);
215 
216    // B: SSE message send → 404 (JSON envelope, pre-stream)
217    const res = await fetch(`${app.baseUrl}/api/v1/workspaces/${workspaceId}/ai/conversations/${mine.id}/messages`, {
218      method: "POST",
219      headers: { "Content-Type": "application/json", Authorization: `Bearer ${b.accessToken}` },
220      body: JSON.stringify({ content: "hi" }),
221    });
222    expect(res.status).toBe(404);
223    const body = (await res.json()) as { error: { code: string } };
224    expect(body.error.code).toBe("NOT_FOUND");
225  });
226 
227  it("non-member gets 403 (JSON envelope)", async () => {
228    const owner = await createUser(app.request, { name: "AI Chat Owner3" });
229    const stranger = await createUser(app.request, { name: "AI Stranger" });
230    const { workspaceId } = await createWorkspace(app.request, authHeaders(owner.accessToken!), "AI Chat WS3");
231    const conv = await createConversation(owner.accessToken!, workspaceId, {});
232    const res = await fetch(`${app.baseUrl}/api/v1/workspaces/${workspaceId}/ai/conversations/${conv.id}/messages`, {
233      method: "POST",
234      headers: { "Content-Type": "application/json", Authorization: `Bearer ${stranger.accessToken}` },
235      body: JSON.stringify({ content: "hi" }),
236    });
237    expect(res.status).toBe(403);
238  });
239});
240 
241describe("TASK-0122 Ask AI (SSE) — stream contract", () => {
242  it("streams: content-type, delta* → usage → done, auto-title, persisted", async () => {
243    const owner = await createUser(app.request, { name: "AI Stream Owner" });
244    const { workspaceId } = await createWorkspace(app.request, authHeaders(owner.accessToken!), "AI Stream WS");
245    const conv = await createConversation(owner.accessToken!, workspaceId, {});
246    expect(conv.id).toBeTruthy();
247 
248    const { status, events, contentType } = await streamMessage(owner.accessToken!, workspaceId, conv.id!, {
249      content: "tell me about quantum gravity",
250    });
251    expect(status).toBe(200);
252    expect(contentType).toContain("text/event-stream");
253 
254    // frame order: deltas first, then usage, then exactly one terminal done
255    const types = events.map((e) => e.type);
256    const firstUsage = types.indexOf("usage");
257    const doneIdx = types.lastIndexOf("done");
258    expect(firstUsage).toBeGreaterThan(-1);
259    expect(doneIdx).toBeGreaterThan(firstUsage);
260    // no non-delta frames before the first usage (keepalives are comments)
261    for (const t of types.slice(0, firstUsage)) expect(t).toBe("delta");
262    expect(types.filter((t) => t === "done")).toHaveLength(1);
263 
264    const usage = events.find((e) => e.type === "usage")!;
265    expect(usage.tokensIn).toBeGreaterThan(0);
266    expect(usage.tokensOut).toBeGreaterThan(0);
267 
268    const done = events.find((e) => e.type === "done")!;
269    expect(done.reason).toBe("ok");
270    expect(done.assistantMessageId).toBeTruthy();
271    expect(done.userMessageId).toBeTruthy();
272 
273    // the mock echoes the user message (word-for-word, first 40 chars)
274    expect(deltaText(events)).toContain("quantum");
275 
276    // persistence: user + assistant rows; assistant carries tokens + model
277    const messages = await app.prisma.aiMessage.findMany({
278      where: { conversationId: conv.id! },
279      orderBy: { createdAt: "asc" },
280    });
281    expect(messages.map((m) => m.role)).toEqual(["user", "assistant"]);
282    expect(messages[1]!.tokensIn).toBe(usage.tokensIn);
283    expect(messages[1]!.tokensOut).toBe(usage.tokensOut);
284    expect(messages[1]!.model).toBeTruthy();
285    expect(messages[1]!.metadata).not.toContain("truncated");
286 
287    // auto-title = first user message (≤60 chars)
288    const convRow = await app.prisma.aiConversation.findUniqueOrThrow({ where: { id: conv.id! } });
289    expect(convRow.title).toBe("tell me about quantum gravity");
290  });
291});
292 
293describe("TASK-0122 Ask AI (SSE) — context binding", () => {
294  it("a question about a page's seeded content answers with that content", async () => {
295    const owner = await createUser(app.request, { name: "AI Ctx Owner" });
296    const { workspaceId } = await createWorkspace(app.request, authHeaders(owner.accessToken!), "AI Ctx WS");
297    const { pageId, blockId } = await createPageWithBlocks(owner.accessToken!, workspaceId, "Context Page", [
298      "The rocket equation governs propulsion.",
299      "Zephyr-9 is the launch site.",
300    ]);
301 
302    const conv = await createConversation(owner.accessToken!, workspaceId, {
303      context: { pageId, blockId },
304    });
305    const { status, events } = await streamMessage(owner.accessToken!, workspaceId, conv.id!, {
306      content: "what is the launch site",
307    });
308    expect(status).toBe(200);
309    // the mock echoes the system-prompt tail → the page content the model SAW
310    const text = deltaText(events);
311    expect(text).toContain("Zephyr-9");
312    expect(text).toContain("Context Page");
313  });
314 
315  it("a user without read access to the referenced page gets context-unavailable, NOT the content", async () => {
316    const owner = await createUser(app.request, { name: "AI CtxDeny Owner" });
317    const member = await createUser(app.request, { name: "AI CtxDeny Member" });
318    const { workspaceId } = await createWorkspace(app.request, authHeaders(owner.accessToken!), "AI CtxDeny WS");
319    await addMemberViaSeam(app.request, app.internalServiceToken, workspaceId, member.email, "MEMBER");
320    const { pageId } = await createPageWithBlocks(owner.accessToken!, workspaceId, "Secret Page", [
321      "The codeword is OPALINE.",
322    ]);
323 
324    // explicit NONE grant on the page for the member (a hard deny —
325    // TASK-0104: overrides the MEMBER role's EDIT baseline)
326    const memberUser = await app.prisma.user.findUniqueOrThrow({ where: { email: member.email } });
327    await app.prisma.permission.create({
328      data: {
329        workspaceId,
330        granteeType: "user",
331        granteeId: memberUser.id,
332        resourceType: "page",
333        resourceId: pageId,
334        level: "NONE",
335      },
336    });
337 
338    const conv = await createConversation(member.accessToken!, workspaceId, { context: { pageId } });
339    const { status, events } = await streamMessage(member.accessToken!, workspaceId, conv.id!, {
340      content: "what is the codeword",
341    });
342    expect(status).toBe(200);
343    const text = deltaText(events);
344    // the unavailable notice is surfaced…
345    expect(text).toMatch(/unavailable/i);
346    // …and the page content is NOT in the prompt (no leak)
347    expect(text).not.toContain("OPALINE");
348    expect(text).not.toContain("Secret Page");
349  });
350});
351 
352describe("TASK-0122 Ask AI (SSE) — budget, retry, rate limit", () => {
353  it("exhausted budget → canonical 402 AI_BUDGET_EXCEEDED (pre-stream JSON)", async () => {
354    const owner = await createUser(app.request, { name: "AI Budget Owner" });
355    const { workspaceId } = await createWorkspace(app.request, authHeaders(owner.accessToken!), "AI Budget WS");
356    // 1-token budget → the first call's estimate (2×maxOutput) blows it
357    await app.prisma.workspace.update({
358      where: { id: workspaceId },
359      data: { usageLimits: { ...(await app.prisma.workspace.findUniqueOrThrow({ where: { id: workspaceId } })).usageLimits, aiMonthlyTokens: 1 } },
360    });
361    const conv = await createConversation(owner.accessToken!, workspaceId, {});
362    const res = await fetch(`${app.baseUrl}/api/v1/workspaces/${workspaceId}/ai/conversations/${conv.id}/messages`, {
363      method: "POST",
364      headers: { "Content-Type": "application/json", Authorization: `Bearer ${owner.accessToken}` },
365      body: JSON.stringify({ content: "hello" }),
366    });
367    expect(res.status).toBe(402);
368    const body = (await res.json()) as { error: { code: string; details?: Array<Record<string, unknown>> } };
369    expect(body.error.code).toBe("AI_BUDGET_EXCEEDED");
370    const detail = body.error.details?.[0] ?? {};
371    expect(detail.limit).toBe(1);
372    // no message rows persisted for the blocked call
373    const messages = await app.prisma.aiMessage.count({ where: { conversationId: conv.id! } });
374    expect(messages).toBe(0);
375  });
376 
377  it("retry (retryOf) re-sends without duplicating the user message row", async () => {
378    const owner = await createUser(app.request, { name: "AI Retry Owner" });
379    const { workspaceId } = await createWorkspace(app.request, authHeaders(owner.accessToken!), "AI Retry WS");
380    const conv = await createConversation(owner.accessToken!, workspaceId, {});
381 
382    const first = await streamMessage(owner.accessToken!, workspaceId, conv.id!, { content: "first message here" });
383    const userMsgId = first.events.find((e) => e.type === "done")!.userMessageId!;
384    expect(userMsgId).toBeTruthy();
385 
386    // simulate a provider outage recovery: the UI's Retry re-sends the SAME
387    // user message (retryOf) — no duplicate user row
388    const second = await streamMessage(owner.accessToken!, workspaceId, conv.id!, {
389      content: "first message here",
390      retryOf: userMsgId,
391    });
392    expect(second.status).toBe(200);
393 
394    const messages = await app.prisma.aiMessage.findMany({
395      where: { conversationId: conv.id! },
396      orderBy: { createdAt: "asc" },
397    });
398    const userRows = messages.filter((m) => m.role === "user");
399    expect(userRows).toHaveLength(1);
400    expect(userRows[0]!.id).toBe(userMsgId);
401    const assistantRows = messages.filter((m) => m.role === "assistant");
402    expect(assistantRows.length).toBeGreaterThanOrEqual(1);
403  });
404 
405  it("per-user 30 msg/min cap → 429 RATE_LIMITED with retryAfterSeconds", async () => {
406    const owner = await createUser(app.request, { name: "AI Rate Owner" });
407    const { workspaceId } = await createWorkspace(app.request, authHeaders(owner.accessToken!), "AI Rate WS");
408    const conv = await createConversation(owner.accessToken!, workspaceId, {});
409 
410    let last: { status: number } = { status: 0 };
411    for (let i = 0; i < 30; i++) {
412      const res = await fetch(`${app.baseUrl}/api/v1/workspaces/${workspaceId}/ai/conversations/${conv.id}/messages`, {
413        method: "POST",
414        headers: { "Content-Type": "application/json", Authorization: `Bearer ${owner.accessToken}` },
415        body: JSON.stringify({ content: `burst ${i}` }),
416      });
417      expect(res.status).toBe(200);
418      await res.body?.cancel().catch(() => {}); // drain not needed — cancel frees the socket
419      last = { status: res.status };
420    }
421    expect(last.status).toBe(200);
422 
423    // 31st → 429 (per-user cap; the per-workspace guard allows 60)
424    const res = await fetch(`${app.baseUrl}/api/v1/workspaces/${workspaceId}/ai/conversations/${conv.id}/messages`, {
425      method: "POST",
426      headers: { "Content-Type": "application/json", Authorization: `Bearer ${owner.accessToken}` },
427      body: JSON.stringify({ content: "burst 30" }),
428    });
429    expect(res.status).toBe(429);
430    const body = (await res.json()) as { error: { code: string; details?: Array<Record<string, unknown>> } };
431    expect(body.error.code).toBe("RATE_LIMITED");
432    expect(typeof body.error.details?.[0]?.retryAfterSeconds).toBe("number");
433  });
434});
435 
436describe("TASK-0122 Ask AI (SSE) — cancellation", () => {
437  it("client abort mid-stream persists a partial assistant message (metadata.truncated)", async () => {
438    const owner = await createUser(app.request, { name: "AI Abort Owner" });
439    const { workspaceId } = await createWorkspace(app.request, authHeaders(owner.accessToken!), "AI Abort WS");
440    const conv = await createConversation(owner.accessToken!, workspaceId, {});
441 
442    // a long user message → many word-deltas (the mock streams one per word);
443    // abort after 3 → the tail is unpersisted.
444    const longMsg = "one two three four five six seven eight nine ten eleven twelve";
445    const { status, events } = await streamMessage(owner.accessToken!, workspaceId, conv.id!, { content: longMsg }, {
446      abortAfterDeltas: 3,
447    });
448    expect(status).toBe(200);
449    const received = deltaText(events);
450    expect(received.length).toBeGreaterThan(0);
451    expect(received).not.toContain("twelve"); // aborted before the tail
452 
453    const messages = await app.prisma.aiMessage.findMany({
454      where: { conversationId: conv.id! },
455      orderBy: { createdAt: "asc" },
456    });
457    // the server persists the partial asynchronously after detecting the
458    // client close — poll briefly for the teardown to land.
459    let assistant = messages.find((m) => m.role === "assistant");
460    for (let i = 0; !assistant && i < 40; i++) {
461      await new Promise((r) => setTimeout(r, 50));
462      const rows = await app.prisma.aiMessage.findMany({ where: { conversationId: conv.id! } });
463      assistant = rows.find((m) => m.role === "assistant");
464    }
465    expect(assistant).toBeTruthy();
466    const meta = assistant!.metadata as { truncated?: boolean } | null;
467    expect(meta?.truncated).toBe(true);
468    expect(assistant!.content.length).toBeLessThan(longMsg.length);
469  });
470});
471