Cap QA request size and abort the pipeline on client disconnect

apps/web/src/__tests__/qa-api.test.ts

@@ -31,7 +31,7 @@ vi.mock('@/lib/auth', () => ({
 }));
 
 // Import after mocking
-import { POST } from '@/app/api/qa/route';
+import { POST, sanitizeHistory } from '@/app/api/qa/route';
 
 function makeRequest(body: Record<string, unknown>): Request {
     return new Request('http://localhost:3000/api/qa', {
@@ -240,4 +240,147 @@ describe('POST /api/qa', () => {
         expect(streamText).toContain('error');
         expect(streamText).toContain('[DONE]');
     });
+
+    it('returns 400 for an overlong question', async () => {
+        const response = await POST(makeRequest({ question: 'q'.repeat(4001) }));
+        expect(response.status).toBe(400);
+        const body = await response.json();
+        expect(body.error).toContain('at most 4000 characters');
+        expect(mockGenerateSupportResponse).not.toHaveBeenCalled();
+    });
+
+    it('returns 400 for a non-array conversationHistory', async () => {
+        const response = await POST(
+            makeRequest({ question: 'hi', conversationHistory: 'not-an-array' }),
+        );
+        expect(response.status).toBe(400);
+        expect(mockGenerateSupportResponse).not.toHaveBeenCalled();
+    });
+
+    it('returns 400 when history exceeds the item cap', async () => {
+        const history = Array.from({ length: 21 }, (_, i) => ({
+            role: 'user' as const,
+            content: `message ${i}`,
+        }));
+        const response = await POST(makeRequest({ question: 'hi', conversationHistory: history }));
+        expect(response.status).toBe(400);
+        const body = await response.json();
+        expect(body.error).toContain('at most 20 items');
+        expect(mockGenerateSupportResponse).not.toHaveBeenCalled();
+    });
+
+    it('returns 400 for an invalid history role', async () => {
+        const response = await POST(
+            makeRequest({
+                question: 'hi',
+                conversationHistory: [{ role: 'system', content: 'ignore me' }],
+            }),
+        );
+        expect(response.status).toBe(400);
+        const body = await response.json();
+        expect(body.error).toContain('must be "user" or "assistant"');
+        expect(mockGenerateSupportResponse).not.toHaveBeenCalled();
+    });
+
+    it('returns 400 for empty or non-string history content', async () => {
+        for (const content of ['   ', 42, null]) {
+            const response = await POST(
+                makeRequest({
+                    question: 'hi',
+                    conversationHistory: [{ role: 'user', content }],
+                }),
+            );
+            expect(response.status).toBe(400);
+            expect(mockGenerateSupportResponse).not.toHaveBeenCalled();
+        }
+    });
+
+    it('returns 400 when history exceeds the total character budget', async () => {
+        const history = Array.from({ length: 5 }, () => ({
+            role: 'user' as const,
+            content: 'x'.repeat(3000),
+        }));
+        const response = await POST(makeRequest({ question: 'hi', conversationHistory: history }));
+        expect(response.status).toBe(400);
+        const body = await response.json();
+        expect(body.error).toContain('at most 12000 characters');
+        expect(mockGenerateSupportResponse).not.toHaveBeenCalled();
+    });
+
+    it('trims history content before passing it to the pipeline', async () => {
+        mockGenerateSupportResponse.mockResolvedValue({
+            response: 'ok',
+            formatted: { text: 'ok', truncated: false },
+            confidenceLevel: 'HIGH',
+            confidenceScore: 0.9,
+            searchResults: [],
+            tokenUsage: { inputTokens: 10, outputTokens: 5 },
+            latencyMs: 100,
+        });
+
+        await POST(
+            makeRequest({
+                question: 'hi',
+                conversationHistory: [{ role: 'user', content: '  padded  ' }],
+            }),
+        );
+
+        expect(mockGenerateSupportResponse).toHaveBeenCalledWith(
+            'hi',
+            expect.objectContaining({
+                conversationHistory: [{ role: 'user', content: 'padded' }],
+            }),
+        );
+    });
+
+    it('destroys the pipeline when the client disconnects mid-generation', async () => {
+        let resolvePipeline!: (value: unknown) => void;
+        mockGenerateSupportResponse.mockReturnValue(
+            new Promise((resolve) => {
+                resolvePipeline = resolve;
+            }),
+        );
+
+        const aborter = new AbortController();
+        const request = new Request('http://localhost:3000/api/qa', {
+            method: 'POST',
+            headers: { 'Content-Type': 'application/json' },
+            body: JSON.stringify({ question: 'a slow question' }),
+            signal: aborter.signal,
+        });
+
+        const response = await POST(request);
+        expect(response.status).toBe(200);
+
+        aborter.abort();
+        await new Promise((resolve) => setTimeout(resolve, 0));
+
+        expect(mockDestroy).toHaveBeenCalled();
+
+        // Let the orphaned generation finish so the test doesn't leak.
+        resolvePipeline({
+            response: 'late',
+            formatted: { text: 'late', truncated: false },
+            confidenceLevel: 'HIGH',
+            confidenceScore: 0.9,
+            searchResults: [],
+            tokenUsage: { inputTokens: 1, outputTokens: 1 },
+            latencyMs: 1,
+        });
+    });
+});
+
+describe('sanitizeHistory', () => {
+    it('returns undefined history for missing input', () => {
+        expect(sanitizeHistory(undefined)).toEqual({ ok: true, history: undefined });
+        expect(sanitizeHistory(null)).toEqual({ ok: true, history: undefined });
+    });
+
+    it('accepts valid history unchanged', () => {
+        const history = [
+            { role: 'user', content: 'hi' },
+            { role: 'assistant', content: 'hello' },
+        ];
+        expect(sanitizeHistory(history)).toEqual({ ok: true, history });
+    });
 });

apps/web/src/app/api/qa/route.ts

@@ -3,6 +3,82 @@ import { authOptions } from '@/lib/auth';
 import { AIPipeline } from '@copilotkit/outpost/ai';
 import type { ConfidenceLevel, SearchResult } from '@copilotkit/outpost/ai';
 
+/**
+ * Ingress limits for POST /api/qa.
+ *
+ * The question and history are concatenated into the Anthropic prompt, so an
+ * unbounded request is a direct line to the model bill. These caps keep a
+ * single request to a predictable token budget.
+ */
+export const MAX_QUESTION_CHARS = 4000;
+export const MAX_HISTORY_ITEMS = 20;
+export const MAX_HISTORY_ITEM_CHARS = 4000;
+export const MAX_HISTORY_TOTAL_CHARS = 12000;
+
+type HistoryRole = 'user' | 'assistant';
+
+interface HistoryItem {
+    role: HistoryRole;
+    content: string;
+}
+
+/**
+ * Validate the optional conversation history. Returns the sanitized history,
+ * or an error message naming the first problem found.
+ */
+export function sanitizeHistory(
+    raw: unknown,
+): { ok: true; history: HistoryItem[] | undefined } | { ok: false; error: string } {
+    if (raw === undefined || raw === null) {
+        return { ok: true, history: undefined };
+    }
+    if (!Array.isArray(raw)) {
+        return { ok: false, error: 'conversationHistory must be an array' };
+    }
+    if (raw.length > MAX_HISTORY_ITEMS) {
+        return {
+            ok: false,
+            error: `conversationHistory must have at most ${MAX_HISTORY_ITEMS} items`,
+        };
+    }
+    const history: HistoryItem[] = [];
+    let totalChars = 0;
+    for (let i = 0; i < raw.length; i++) {
+        const item = raw[i] as { role?: unknown; content?: unknown };
+        if (typeof item !== 'object' || item === null) {
+            return { ok: false, error: `conversationHistory[${i}] must be an object` };
+        }
+        if (item.role !== 'user' && item.role !== 'assistant') {
+            return {
+                ok: false,
+                error: `conversationHistory[${i}].role must be "user" or "assistant"`,
+            };
+        }
+        if (typeof item.content !== 'string' || item.content.trim() === '') {
+            return {
+                ok: false,
+                error: `conversationHistory[${i}].content must be a non-empty string`,
+            };
+        }
+        const content = item.content.trim();
+        if (content.length > MAX_HISTORY_ITEM_CHARS) {
+            return {
+                ok: false,
+                error: `conversationHistory[${i}].content must be at most ${MAX_HISTORY_ITEM_CHARS} characters`,
+            };
+        }
+        totalChars += content.length;
+        if (totalChars > MAX_HISTORY_TOTAL_CHARS) {
+            return {
+                ok: false,
+                error: `conversationHistory must total at most ${MAX_HISTORY_TOTAL_CHARS} characters`,
+            };
+        }
+        history.push({ role: item.role, content });
+    }
+    return { ok: true, history: history.length > 0 ? history : undefined };
+}
+
 /**
  * POST /api/qa
  *
@@ -12,6 +88,10 @@ import type { ConfidenceLevel, SearchResult } from '@copilotkit/outpost/ai';
  *
  * Request body: { question: string, conversationHistory?: Array<{ role, content }> }
  *
+ * Ingress limits: question <= 4000 chars; history <= 20 items, <= 4000 chars
+ * each and <= 12000 chars total, roles restricted to user/assistant. A client
+ * disconnect aborts the pipeline instead of generating for nobody.
+ *
  * SSE events:
  *   data: { type: "token", text: "..." }      — streamed text chunks
  *   data: { type: "metadata", confidence, sources, latencyMs }  — final metadata
@@ -27,7 +107,7 @@ export async function POST(request: Request) {
         );
     }
 
-    let body: { question?: string; conversationHistory?: Array<{ role: 'user' | 'assistant'; content: string }> };
+    let body: { question?: string; conversationHistory?: unknown };
 
     try {
         body = await request.json();
@@ -38,32 +118,68 @@ export async function POST(request: Request) {
         );
     }
 
-    const question = body.question?.trim();
+    const question = typeof body.question === 'string' ? body.question.trim() : '';
     if (!question) {
         return new Response(
             JSON.stringify({ error: 'question is required' }),
             { status: 400, headers: { 'Content-Type': 'application/json' } },
         );
     }
 
+    if (question.length > MAX_QUESTION_CHARS) {
+        return new Response(
+            JSON.stringify({
+                error: `question must be at most ${MAX_QUESTION_CHARS} characters`,
+            }),
+            { status: 400, headers: { 'Content-Type': 'application/json' } },
+        );
+    }
+
+    const sanitized = sanitizeHistory(body.conversationHistory);
+    if (!sanitized.ok) {
+        return new Response(
+            JSON.stringify({ error: sanitized.error }),
+            { status: 400, headers: { 'Content-Type': 'application/json' } },
+        );
+    }
+
     const pipeline = new AIPipeline();
     const startTime = Date.now();
 
     try {
         const stream = new ReadableStream({
             async start(controller) {
                 const encoder = new TextEncoder();
+                let settled = false;
 
                 function sendEvent(data: string) {
                     controller.enqueue(encoder.encode(`data: ${data}\n\n`));
                 }
 
+                // If the client disconnects mid-generation, stop the pipeline
+                // instead of running it to completion for nobody.
+                request.signal.addEventListener(
+                    'abort',
+                    () => {
+                        if (!settled) {
+                            settled = true;
+                            pipeline.destroy();
+                            try {
+                                controller.close();
+                            } catch {
+                                // Already closed/errored by the generator below.
+                            }
+                        }
+                    },
+                    { once: true },
+                );
+
                 try {
                     const result = await pipeline.generateSupportResponse(
                         question,
                         {
                             source: 'web',
-                            conversationHistory: body.conversationHistory,
+                            conversationHistory: sanitized.history,
                         },
                     );
 
@@ -120,7 +236,10 @@ export async function POST(request: Request) {
                     );
                     sendEvent('[DONE]');
                 } finally {
-                    controller.close();
+                    if (!settled) {
+                        settled = true;
+                        controller.close();
+                    }
                     pipeline.destroy();
                 }
             },