feat(ai): server-owned AI streams — clients become pure subscribers (core)

apps/web/eslint.config.mjs

@@ -172,6 +172,41 @@ const eslintConfig = [
       ],
     },
   },
+  {
+    // AI stream surfaces: a stale closure here is not a re-render bug, it is a RUNAWAY AGENT.
+    //
+    // Streams are server-owned and deliberately survive a client disconnect, so the only thing
+    // that stops a generation is an explicit abort naming it correctly. Every stale value in
+    // these files — a captured conversationId, a captured messageId, a captured isStreaming —
+    // means Stop names the wrong stream (or nothing), the fetch is cancelled, the button flips
+    // back to Send, and the server keeps generating, keeps running write tools, and KEEPS
+    // BILLING while the user believes it stopped. That has now happened repeatedly.
+    //
+    // `react-hooks/exhaustive-deps` catches exactly this, and it was already enabled — as a
+    // WARNING, via next/core-web-vitals. `bun run lint` exits 0 on warnings, so CI reported
+    // "14/14 successful" while the rule was pointing straight at a missing dep in handleStop.
+    // The signal existed and was wired to nothing.
+    //
+    // Scoped to the files that own stream identity, and verified to be at ZERO violations and
+    // ZERO suppressions when added — so this costs nothing today and fails CI on the next one.
+    // If it fires, do not silence it: a dep you are tempted to omit here is a stop button you
+    // are about to break.
+    files: [
+      "src/hooks/useChannelStreamSocket.ts",
+      "src/hooks/useAgentChannelMultiplayer.ts",
+      "src/hooks/useAppStateRecovery.ts",
+      "src/contexts/GlobalChatContext.tsx",
+      "src/components/layout/middle-content/page-views/dashboard/GlobalAssistantView.tsx",
+      "src/components/layout/middle-content/page-views/dashboard/useGlobalEffectiveStream.ts",
+      "src/components/layout/middle-content/page-views/ai-page/AiChatView.tsx",
+      "src/components/layout/right-sidebar/ai-assistant/SidebarChatTab.tsx",
+      "src/lib/ai/shared/hooks/useChatStop.ts",
+      "src/lib/ai/shared/hooks/useChatTransport.ts",
+    ],
+    rules: {
+      "react-hooks/exhaustive-deps": "error",
+    },
+  },
 ];
 
 export default eslintConfig;

apps/web/src/app/api/ai/abort/__tests__/route.test.ts

@@ -6,6 +6,10 @@
  */
 
 import { describe, it, expect, beforeEach, vi } from 'vitest';
+
+const mockAbortConversationStreams = vi.hoisted(() =>
+  vi.fn().mockResolvedValue({ aborted: [], reason: 'No in-flight stream on this conversation' }),
+);
 import { NextResponse } from 'next/server';
 import { POST } from '../route';
 import type { SessionAuthResult, AuthError } from '@/lib/auth';
@@ -17,6 +21,10 @@ vi.mock('@/lib/auth', () => ({
 }));
 
 // Mock stream abort registry (boundary)
+vi.mock('@/lib/ai/core/abort-conversation-streams', () => ({
+  abortConversationStreams: mockAbortConversationStreams,
+}));
+
 vi.mock('@/lib/ai/core/stream-abort-registry', () => ({
   abortStream: vi.fn(),
   abortStreamByMessageId: vi.fn(),
@@ -96,6 +104,59 @@ describe('POST /api/ai/abort', () => {
     });
   });
 
+  // THE WINDOW STOP COULD NOT NAME.
+  //
+  // streamId and messageId are BOTH minted server-side, and the client learns neither until the
+  // response headers land. A real agent send spends 0.5-3s before that (auth, rate limit, DB
+  // reads, context assembly, provider connect). Press Stop in that window — exactly when a user
+  // who spotted a typo presses it — and the client had nothing to name: the abort was a
+  // guaranteed no-op, the fetch was cancelled, and the button flipped back to Send.
+  //
+  // Cancelling the fetch stops NOTHING: streams are deliberately server-owned and survive a
+  // client disconnect. The generation kept running, kept calling write tools, and kept BILLING,
+  // while the UI said it had stopped. conversationId is the one name the client holds from t=0.
+  describe('abort by conversationId (the pre-headers window)', () => {
+    it('accepts conversationId alone and aborts the conversation\'s in-flight stream', async () => {
+      vi.mocked(authenticateRequestWithOptions).mockResolvedValueOnce(mockWebAuth(mockUserId));
+      vi.mocked(isAuthError).mockReturnValueOnce(false);
+      mockAbortConversationStreams.mockResolvedValueOnce({ aborted: ['msg-1'], reason: '' });
+
+      const response = await POST(createRequest({ conversationId: 'conv-1' }));
+      const body = await response.json();
+
+      expect(response.status).toBe(200);
+      expect(mockAbortConversationStreams).toHaveBeenCalledWith({
+        conversationId: 'conv-1',
+        userId: expect.any(String),
+      });
+      expect(body.aborted).toBe(true);
+    });
+
+    it('given nothing was in flight, reports honestly rather than claiming success', async () => {
+      vi.mocked(authenticateRequestWithOptions).mockResolvedValueOnce(mockWebAuth(mockUserId));
+      vi.mocked(isAuthError).mockReturnValueOnce(false);
+      mockAbortConversationStreams.mockResolvedValueOnce({
+        aborted: [],
+        reason: 'No in-flight stream on this conversation',
+      });
+
+      const body = await (await POST(createRequest({ conversationId: 'conv-1' }))).json();
+
+      expect(body.aborted).toBe(false);
+    });
+
+    // messageId names the stream exactly; conversationId is only the fallback for when nothing
+    // else exists yet. Preferring the fallback would abort every stream on the conversation.
+    it('prefers messageId when both are supplied', async () => {
+      vi.mocked(authenticateRequestWithOptions).mockResolvedValueOnce(mockWebAuth(mockUserId));
+      vi.mocked(isAuthError).mockReturnValueOnce(false);
+
+      await POST(createRequest({ messageId: 'msg-precise', conversationId: 'conv-1' }));
+
+      expect(mockAbortConversationStreams).not.toHaveBeenCalled();
+    });
+  });
+
   describe('Input Validation', () => {
     it('returns 400 when neither streamId nor messageId provided', async () => {
       vi.mocked(authenticateRequestWithOptions).mockResolvedValueOnce(mockWebAuth(mockUserId));
@@ -106,7 +167,7 @@ describe('POST /api/ai/abort', () => {
       const data = await response.json();
 
       expect(response.status).toBe(400);
-      expect(data.error).toBe('streamId or messageId is required');
+      expect(data.error).toBe('streamId, messageId or conversationId is required');
       expect(abortStream).not.toHaveBeenCalled();
       expect(abortStreamByMessageId).not.toHaveBeenCalled();
     });
@@ -120,7 +181,7 @@ describe('POST /api/ai/abort', () => {
       const data = await response.json();
 
       expect(response.status).toBe(400);
-      expect(data.error).toBe('streamId or messageId is required');
+      expect(data.error).toBe('streamId, messageId or conversationId is required');
     });
 
     it('returns 400 when streamId is empty string and no messageId', async () => {
@@ -132,7 +193,7 @@ describe('POST /api/ai/abort', () => {
       const data = await response.json();
 
       expect(response.status).toBe(400);
-      expect(data.error).toBe('streamId or messageId is required');
+      expect(data.error).toBe('streamId, messageId or conversationId is required');
     });
 
     it('returns 400 when streamId is whitespace only and no messageId', async () => {
@@ -144,7 +205,7 @@ describe('POST /api/ai/abort', () => {
       const data = await response.json();
 
       expect(response.status).toBe(400);
-      expect(data.error).toBe('streamId or messageId is required');
+      expect(data.error).toBe('streamId, messageId or conversationId is required');
       expect(abortStream).not.toHaveBeenCalled();
     });
   });
@@ -249,7 +310,7 @@ describe('POST /api/ai/abort', () => {
       const data = await response.json();
 
       expect(response.status).toBe(400);
-      expect(data.error).toBe('streamId or messageId is required');
+      expect(data.error).toBe('streamId, messageId or conversationId is required');
     });
   });
 

apps/web/src/app/api/ai/abort/route.ts

@@ -1,6 +1,7 @@
 import { NextResponse } from 'next/server';
 import { authenticateRequestWithOptions, isAuthError } from '@/lib/auth';
 import { abortStream, abortStreamByMessageId } from '@/lib/ai/core/stream-abort-registry';
+import { abortConversationStreams } from '@/lib/ai/core/abort-conversation-streams';
 import { loggers } from '@pagespace/lib/logging/logger-config';
 import { auditRequest } from '@pagespace/lib/audit/audit-log';
 import { checkDistributedRateLimit } from '@pagespace/lib/security/distributed-rate-limit';
@@ -48,23 +49,44 @@ export async function POST(request: Request) {
     }
 
     const body = await request.json();
-    const { streamId, messageId } = body as { streamId?: string; messageId?: string };
+    const { streamId, messageId, conversationId } = body as {
+      streamId?: string;
+      messageId?: string;
+      conversationId?: string;
+    };
 
     const hasStreamId = streamId && typeof streamId === 'string' && streamId.trim();
     const hasMessageId = messageId && typeof messageId === 'string' && messageId.trim();
+    // The one name the client holds from t=0. Both streamId and messageId are minted SERVER-side
+    // and are unknown until the response headers land — so a Stop pressed inside the 0.5-3s TTFB
+    // window (auth, rate limit, DB reads, context assembly, provider connect) had NOTHING to
+    // name. It cancelled the local fetch and returned; streams are deliberately server-owned and
+    // survive a client disconnect, so the generation kept running, kept calling write tools, and
+    // kept billing, while the button flipped back to Send. See abort-conversation-streams.ts.
+    const hasConversationId =
+      conversationId && typeof conversationId === 'string' && conversationId.trim();
 
-    if (!hasStreamId && !hasMessageId) {
+    if (!hasStreamId && !hasMessageId && !hasConversationId) {
       return NextResponse.json(
-        { error: 'streamId or messageId is required' },
+        { error: 'streamId, messageId or conversationId is required' },
         { status: 400 }
       );
     }
 
+    // messageId is the most precise name, then streamId; conversationId is the fallback that
+    // works before either exists.
     const result = hasMessageId
       ? abortStreamByMessageId({ messageId: messageId as string, userId })
-      : abortStream({ streamId: streamId as string, userId });
+      : hasStreamId
+        ? abortStream({ streamId: streamId as string, userId })
+        : await abortConversationStreams({ conversationId: conversationId as string, userId })
+            .then((r) => ({ aborted: r.aborted.length > 0, reason: r.reason }));
 
-    const resourceId = hasMessageId ? (messageId as string) : (streamId as string);
+    const resourceId = hasMessageId
+      ? (messageId as string)
+      : hasStreamId
+        ? (streamId as string)
+        : (conversationId as string);
 
     loggers.api.info('AI stream abort requested', {
       streamId,