ai-rag.integration.test.ts

ai-rag.integration.test.tstests/integration/ai-rag.integration.test.tsraw ← integration16,520 B← Back to the app
1/**
2 * TASK-0125 — RAG workspace search (query path) integration suite.
3 *
4 * Full stack (AppModule + Postgres + Redis) + the shared OpenSearch content
5 * index, against the deterministic `MockProvider` (`AI_PROVIDER=mock`; the
6 * `rag` mode emits a cited answer derived from the numbered context). The
7 * query path is the spec's proof surface:
8 *
9 *   - the FIRST SSE frame is `sources` (the server-side citations, built from
10 *     the Phase-8 hybrid retrieval — BM25 + kNN + RRF, both legs filtered by
11 *     the requester's tenant + visibility snapshot);
12 *   - permission isolation: a non-member workspace → 404; a member without
13 *     access to the page → ZERO citations for it, both via the RAG endpoint
14 *     AND via the raw hybrid search (mirrors the Phase-8 proof);
15 *   - hybrid determinism: the same query twice → the same citation order;
16 *   - persistence: the answer row is hidden (never in the conversation list),
17 *     carries `metadata.feature="rag"` + the citation list in `toolResults`.
18 *
19 * Indexing is seeded DIRECTLY (SearchClient.indexDocs with the mock vector) —
20 * the chunk+embed WORKER pipeline is covered by the worker unit suite
21 * (`ai-embedding.test.ts`); the worker's event-driven indexing is E2E-only
22 * (it needs the domain-event stream, which the Testcontainers app does not
23 * emit here). Recorded decision: this spec proves the QUERY path end-to-end.
24 *
25 * ── RUNNER GATE (Docker + OpenSearch) — SELF-SKIPS WHEN THE STACK IS ABSENT ──
26 * `dockerAvailable() && opensearchUp()` (the API reads the SAME OpenSearch
27 * the worker writes), so an absent stack never fails the suite.
28 */
29// Config must be fixed BEFORE the app boots (loadEnv + the workspace cache).
30process.env.AI_PROVIDER = "mock";
31process.env.SEARCH_VECTOR_DIMENSION = "8";
32process.env.AI_MOCK_DELTA_DELAY_MS = "5";
33 
34import { afterAll, beforeAll, describe, expect, it } from "vitest";
35import { execSync } from "node:child_process";
36import {
37  SearchClient,
38  contentMapping,
39  type GrantSnapshot,
40  type SearchDoc,
41} from "@factory/search";
42import { createTestApp, type TestAppHandle } from "./helpers/api";
43import { addMemberViaSeam, authHeaders, createWorkspace, createUser } from "./helpers/seed";
44 
45/** The OpenSearch the API talks to (its SEARCH_URL default). */
46const SEARCH_URL = process.env.SEARCH_URL ?? "http://localhost:9200";
47const CONTENT_INDEX = "factory-content";
48const CONTENT_INDEX_ALIAS = "factory-content";
49 
50/** A distinctive, unique seeded fact (the spec's "Zanzibar-7"). */
51const FACT = "The launch code word is Zanzibar-7.";
52const QUERY = "What is the launch code word?";
53 
54function dockerAvailable(): boolean {
55  try {
56    execSync("docker info", { stdio: "ignore", timeout: 8_000 });
57    return true;
58  } catch {
59    return false;
60  }
61}
62async function opensearchUp(url: string): Promise<boolean> {
63  try {
64    const res = await fetch(`${url.replace(/\/+$/, "")}/_cluster/health`, { signal: AbortSignal.timeout(4_000) });
65    return res.ok;
66  } catch {
67    return false;
68  }
69}
70const stackReady = dockerAvailable() && (await opensearchUp(SEARCH_URL));
71 
72let app: TestAppHandle;
73let client: SearchClient;
74// A run-unique suffix so seeded docs never collide across runs in the shared index.
75const RUN = `rag${Date.now().toString(36)}`;
76 
77// Fixtures.
78let wsA = ""; // A's workspace (the page's tenant).
79let pageA = ""; // the seeded page id (A owns it).
80let userA: Awaited<ReturnType<typeof createUser>> | null = null;
81let userB: Awaited<ReturnType<typeof createUser>> | null = null;
82let userC: Awaited<ReturnType<typeof createUser>> | null = null;
83let pageDocId = ""; // `page:<id>`
84let chunkDocId = ""; // `page:<id>#c0`
85 
86/** One SSE frame (the RAG wire contract). */
87type RagSseEvent = {
88  type: string;
89  sources?: Array<Record<string, unknown>>;
90  text?: string;
91  tokensIn?: number;
92  tokensOut?: number;
93  code?: string;
94  message?: string;
95  assistantMessageId?: string;
96  userMessageId?: string;
97  reason?: string;
98};
99 
100/** POST the RAG endpoint, collect the SSE frames. */
101async function ragStream(accessToken: string, body: Record<string, unknown>): Promise<{ status: number; events: RagSseEvent[]; contentType: string }> {
102  const res = await fetch(`${app.baseUrl}/api/v1/workspaces/${wsA}/ai/search`, {
103    method: "POST",
104    headers: { "Content-Type": "application/json", Authorization: `Bearer ${accessToken}` },
105    body: JSON.stringify(body),
106  });
107  const contentType = res.headers.get("content-type") ?? "";
108  const events: RagSseEvent[] = [];
109  if (!res.body) return { status: res.status, events, contentType };
110  const reader = res.body.getReader();
111  const decoder = new TextDecoder();
112  let buf = "";
113  for (;;) {
114    const chunk = await reader.read();
115    if (chunk.done) break;
116    buf += decoder.decode(chunk.value, { stream: true });
117    let idx: number;
118    while ((idx = buf.indexOf("\n\n")) !== -1) {
119      const frame = buf.slice(0, idx);
120      buf = buf.slice(idx + 2);
121      for (const line of frame.split("\n")) {
122        if (!line.startsWith("data:")) continue; // skip `: keepalive`
123        try {
124          events.push(JSON.parse(line.slice(5).trim()) as RagSseEvent);
125        } catch {
126          /* malformed frame — skip */
127        }
128      }
129    }
130  }
131  return { status: res.status, events, contentType };
132}
133 
134/** The deterministic mock vector for a text (dim 8): v0=len/100 (≤1), v1=0.5. */
135function mockVector(text: string, dim = 8): number[] {
136  const v = new Array<number>(dim).fill(0);
137  v[0] = Math.min(1, text.length / 100);
138  if (dim > 1) v[1] = 0.5;
139  return v;
140}
141 
142/**
143 * The page's grant snapshot, reproducing the worker's output for a PRIVATE
144 * page (grants.ts `workspace = !isPrivate && …` → false; `users = [creator]`):
145 * ONLY the creator (A) can see it — a workspace member (B) without a page
146 * grant is excluded. This is the permission-isolation scenario: B is a
147 * member of the workspace but has NO access to page P.
148 */
149function pageGrants(creatorId: string): GrantSnapshot {
150  return { public: false, workspace: false, user: true, teams: [], users: [creatorId] };
151}
152 
153beforeAll(async () => {
154  app = await createTestApp();
155  client = new SearchClient({ baseUrl: SEARCH_URL });
156 
157  // Ensure the content index (or the alias to it) exists with the canonical
158  // mapping. The dev stack rolled it to `factory-content-v2` behind the alias;
159  // a fresh stack has nothing. `indexExists` resolves through the alias, so
160  // this is idempotent for BOTH the real-index and the alias cases.
161  if (!(await client.indexExists(CONTENT_INDEX_ALIAS))) {
162    const mapping = contentMapping();
163    await client.createIndexIfMissing(`${CONTENT_INDEX}-${RUN}`, {
164      settings: mapping.settings as Record<string, unknown>,
165      mappings: mapping.mappings as Record<string, unknown>,
166    });
167    await client.createAlias(`${CONTENT_INDEX}-${RUN}`, CONTENT_INDEX_ALIAS);
168  }
169 
170  // ---- fixtures through the API (real authorization) ----
171  userA = await createUser(app.request, { name: "RAG Owner A" });
172  wsA = (await createWorkspace(app.request, authHeaders(userA.accessToken!), `RAG WS A ${RUN}`)).workspaceId;
173  userB = await createUser(app.request, { name: "RAG Member B" });
174  userC = await createUser(app.request, { name: "RAG Outsider C" });
175  await addMemberViaSeam(app.request, app.internalServiceToken, wsA, userB.email, "MEMBER");
176 
177  // The seeded page (A owns it; B is a workspace member but has NO page-level
178  // grant — a page-level privacy scenario, so B's snapshot excludes it).
179  const pageRes = await app.request("POST", `/api/v1/workspaces/${wsA}/pages`, {
180    headers: authHeaders(userA.accessToken!),
181    body: { title: "Launch Runbook" },
182  });
183  expect(pageRes.status).toBe(201);
184  pageA = (await pageRes.json()) as { id: string };
185  const blockRes = await app.request("POST", `/api/v1/pages/${pageA.id}/blocks`, {
186    headers: authHeaders(userA.accessToken!),
187    body: { type: "PARAGRAPH", content: { text: FACT } },
188  });
189  expect(blockRes.status).toBe(201);
190 
191  // Seed the index DIRECTLY: the whole page doc + one chunk doc (the query
192  // path de-dupes chunks to the parent). Both carry the mock vector (dim 8).
193  const body = `Launch Runbook\n${FACT}`;
194  const vec = mockVector(body);
195  pageDocId = `page:${pageA.id}`;
196  chunkDocId = `${pageDocId}#c0`;
197  const base = {
198    workspaceId: wsA,
199    title: "Launch Runbook",
200    pageId: pageA.id,
201    url: `/w/${wsA}/page/${pageA.id}`,
202    grants: pageGrants(userA.userId),
203    archived: false,
204    createdAt: "2025-01-01T00:00:00.000Z",
205    updatedAt: "2025-06-01T00:00:00.000Z",
206    vectorModel: "mock-embed",
207    vector: vec,
208  };
209  await client.indexDocs([
210    { ...base, docId: pageDocId, type: "page", id: pageA.id, body } satisfies SearchDoc,
211    {
212      ...base,
213      docId: chunkDocId,
214      type: "chunk",
215      id: pageA.id,
216      body: FACT,
217      chunkId: "c0",
218      chunkIndex: 0,
219      parentDocId: pageDocId,
220    } satisfies SearchDoc,
221  ]);
222  await client.refresh(CONTENT_INDEX_ALIAS);
223 
224  // The registry parity row (what the worker would write) — the query path
225  // does not read it, but the test asserts the count equality the spec calls
226  // for (OpenSearch chunk docs vs ai_embeddings rows for the workspace).
227  await app.prisma.aiEmbedding.create({
228    data: {
229      workspaceId: wsA,
230      chunkRef: chunkDocId,
231      chunkText: FACT,
232      vectorId: chunkDocId,
233      model: "mock-embed",
234      dimension: 8,
235      metadata: { docType: "page", refId: pageA.id, index: CONTENT_INDEX_ALIAS },
236    },
237  });
238}, 180_000);
239 
240afterAll(async () => {
241  // Best-effort cleanup of the seeded docs (the shared index persists).
242  try {
243    await (
244      await fetch(`${SEARCH_URL}/${CONTENT_INDEX_ALIAS}/_delete_by_query`, {
245        method: "POST",
246        headers: { "Content-Type": "application/json" },
247        body: JSON.stringify({ query: { term: { pageId: pageA } } }),
248        refresh: true,
249      })
250    ).json();
251    await app.prisma.aiEmbedding.deleteMany({ where: { workspaceId: wsA } });
252  } catch {
253    /* shared index — do not fail teardown */
254  }
255  await app?.close();
256});
257 
258describe.skipIf(!stackReady)("RAG workspace search (TASK-0125 query path)", () => {
259  it("happy path: sources-first SSE, the cited answer contains the fact + [n], usage + done ok", async () => {
260    const { status, events, contentType } = await ragStream(userA!.accessToken!, { query: QUERY });
261    expect(status).toBe(200);
262    expect(contentType).toContain("text/event-stream");
263 
264    // The FIRST data frame is `sources` (the server-side citations).
265    expect(events[0]?.type).toBe("sources");
266    const sources = events[0]?.sources ?? [];
267    expect(sources.length).toBeGreaterThan(0);
268    const cited = sources.find((s) => s.docId === pageDocId);
269    expect(cited, "the seeded page must be cited").toBeTruthy();
270    expect(cited?.title).toBe("Launch Runbook");
271    expect(cited?.type).toBe("page");
272    expect(cited?.n).toBe(1);
273    // Chunk hits de-dup to the parent → the page is cited ONCE.
274    expect(sources.filter((s) => s.id === pageA.id).length).toBe(1);
275 
276    // The answer streams, contains the seeded fact + the citation marker.
277    const answer = events.filter((e) => e.type === "delta").map((e) => e.text ?? "").join("");
278    expect(answer).toContain("Zanzibar-7");
279    expect(answer).toContain("[1]");
280 
281    // usage + a terminal done ok.
282    expect(events.some((e) => e.type === "usage")).toBe(true);
283    const done = events.find((e) => e.type === "done");
284    expect(done?.reason).toBe("ok");
285    expect(done?.assistantMessageId).toBeTruthy();
286    expect(done?.userMessageId).toBeTruthy();
287  });
288 
289  it("non-member workspace → 404 (the search viewer's fail-closed rule)", async () => {
290    const res = await fetch(`${app.baseUrl}/api/v1/workspaces/${wsA}/ai/search`, {
291      method: "POST",
292      headers: { "Content-Type": "application/json", Authorization: `Bearer ${userC!.accessToken}` },
293      body: JSON.stringify({ query: QUERY }),
294    });
295    expect(res.status).toBe(404);
296    const body = (await res.json()) as { error?: { code?: string } };
297    expect(body.error?.code).toBe("NOT_FOUND");
298  });
299 
300  it("member without page access → ZERO citations for it (RAG + raw hybrid both prove the permission filter)", async () => {
301    const { status, events } = await ragStream(userB!.accessToken!, { query: QUERY });
302    expect(status).toBe(200);
303    const sources = events.find((e) => e.type === "sources")?.sources ?? [];
304    // B is a workspace member, but the page's snapshot does not include B →
305    // the page is invisible to B's viewer → zero citations for it.
306    expect(sources.find((s) => s.id === pageA)).toBeUndefined();
307    // The no-results answer is streamed (the mock's deterministic empty text).
308    const answer = events.filter((e) => e.type === "delta").map((e) => e.text ?? "").join("");
309    expect(answer).toContain("I could not find any workspace content");
310 
311    // Raw search (the Phase-8 proof, mirrored): the same query through the
312    // public /search endpoint (the viewer's tenant + visibility filter on the
313    // keyword leg — and the RAG hybrid's kNN leg uses the SAME visibility
314    // filter) also returns NO page doc for B.
315    const searchRes = await fetch(`${app.baseUrl}/api/v1/search`, {
316      method: "POST",
317      headers: { "Content-Type": "application/json", Authorization: `Bearer ${userB!.accessToken}` },
318      body: JSON.stringify({ q: QUERY }),
319    });
320    expect(searchRes.status).toBe(200);
321    const searchBody = (await searchRes.json()) as { data?: Array<Record<string, unknown>> };
322    const data = searchBody.data ?? [];
323    expect(data.find((d) => d.id === pageA || d.pageId === pageA), "B's raw search must not surface the page").toBeUndefined();
324  }, 60_000);
325 
326  it("hybrid determinism: the same query twice → the same citation order", async () => {
327    const a = (await ragStream(userA!.accessToken!, { query: QUERY })).events.find((e) => e.type === "sources")?.sources ?? [];
328    const b = (await ragStream(userA!.accessToken!, { query: QUERY })).events.find((e) => e.type === "sources")?.sources ?? [];
329    expect(b.map((s) => `${s.docId}:${s.n}`)).toEqual(a.map((s) => `${s.docId}:${s.n}`));
330    expect(a.length).toBeGreaterThan(0);
331  });
332 
333  it("persistence: the answer row is hidden + carries feature=rag + the citations in toolResults", async () => {
334    const { events } = await ragStream(userA!.accessToken!, { query: QUERY });
335    const done = events.find((e) => e.type === "done")!;
336    const row = await app.prisma.aiMessage.findUniqueOrThrow({ where: { id: done.assistantMessageId! } });
337    expect(row.role).toBe("assistant");
338    expect(row.metadata).toMatchObject({ feature: "rag" });
339    const toolResults = (row.toolResults ?? {}) as { sources?: Array<{ id?: string }> };
340    expect(toolResults.sources?.some((s) => s.id === pageA.id)).toBe(true);
341 
342    // The hidden per-call conversation never appears in the panel list.
343    const convRes = await app.request("GET", `/api/v1/workspaces/${wsA}/ai/conversations`, {
344      headers: authHeaders(userA.accessToken!),
345    });
346    expect(convRes.status).toBe(200);
347    const convs = ((await convRes.json()) as { data: Array<{ title: string | null }> }).data;
348    expect(convs.some((c) => (c.title ?? "").startsWith("rag:"))).toBe(false);
349  });
350 
351  it("rate limit: the 31st rapid RAG query in the per-user window → 429", async () => {
352    // A workspace MEMBER (so the 404 non-member gate does not fire first) —
353    // the shared ai.calls limiter (30/min) is the gate under test.
354    const userD = await createUser(app.request, { name: `RAG Rate D ${RUN}` });
355    await addMemberViaSeam(app.request, app.internalServiceToken, wsA, userD.email, "MEMBER");
356    let saw429 = false;
357    let lastStatus = 0;
358    for (let i = 0; i < 31; i++) {
359      const res = await fetch(`${app.baseUrl}/api/v1/workspaces/${wsA}/ai/search`, {
360        method: "POST",
361        headers: { "Content-Type": "application/json", Authorization: `Bearer ${userD.accessToken}` },
362        body: JSON.stringify({ query: QUERY }),
363      });
364      lastStatus = res.status;
365      if (res.status === 429) {
366        saw429 = true;
367        const body = (await res.json()) as { error?: { code?: string } };
368        expect(body.error?.code).toBe("RATE_LIMITED");
369        break;
370      }
371      // Drain the (short) stream so the next request starts clean.
372      if (res.body) await res.body.cancel();
373    }
374    expect(saw429, "a 429 must fire within 31 rapid queries").toBe(true);
375    expect(lastStatus).toBe(429);
376  }, 120_000);
377});
378