feat(ai): extend the injection guard to the stream

.eslintrc.js

@@ -4,6 +4,13 @@ module.exports = {
     'node': true,
     'jest': true
   },
+  globals: {
+    /**
+     * Global since Node 15 (this project runs Node 24 per .nvmrc), but not part of
+     * eslint's "node" env, which predates it
+     */
+    'AbortController': 'readonly'
+  },
   rules: {
     '@typescript-eslint/camelcase': 'warn',
     '@typescript-eslint/no-unused-vars': 'warn',

package.json

@@ -1,6 +1,6 @@
 {
   "name": "hawk.api",
-  "version": "1.5.13",
+  "version": "1.5.15",
   "main": "index.ts",
   "license": "BUSL-1.1",
   "scripts": {

src/directives/requireUserInWorkspace.ts

@@ -37,7 +37,7 @@ async function checkUserInWorkspaceByWorkspaceId(context: ResolverContextBase, w
  * @param context - request context
  * @param projectId - project id
  */
-async function checkUserInWorkspaceByProjectId(context: ResolverContextBase, projectId: string): Promise<void> {
+export async function checkUserInWorkspaceByProjectId(context: ResolverContextBase, projectId: string): Promise<void> {
   const userId = context.user.id;
 
   if (userId) {

src/index.ts

@@ -32,6 +32,7 @@ import ReleasesFactory from './models/releasesFactory';
 import RedisHelper from './redisHelper';
 import { appendSsoRoutes } from './sso';
 import { appendGitHubRoutes } from './integrations/github';
+import { appendAiAssistantRoutes } from './services/askAi';
 
 /**
  * Option to enable playground
@@ -272,6 +273,11 @@ class HawkAPI {
      */
     appendGitHubRoutes(this.app, sharedFactories);
 
+    /**
+     * Append AI assistant route to Express app
+     */
+    appendAiAssistantRoutes(this.app);
+
     await this.server.start();
     this.app.use(graphqlUploadExpress());
     this.server.applyMiddleware({ app: this.app });

src/integrations/vercel-ai/index.ts

@@ -1,4 +1,6 @@
-import { generateText } from 'ai';
+import { generateText, streamText, type TextStreamPart, type ToolSet } from 'ai';
+import { getErrorMessage, ProviderOptions } from '@ai-sdk/provider-utils';
+import type { SuggestionStream } from '../../services/askAi/suggestionStream';
 
 /**
  * Params for a single completion call to the model
@@ -15,6 +17,43 @@ export interface CompletionParams {
   prompt: string;
 }
 
+/**
+ * Params for a streaming completion call to the model
+ */
+export interface StreamParams extends CompletionParams {
+  /**
+   * Aborted when the answer is no longer wanted, which stops the model
+   */
+  signal: AbortSignal;
+}
+
+/**
+ * Translate the SDK's stream parts into suggestion parts, dropping the ones
+ * nothing reads.
+ *
+ * @param parts - stream of parts in the shape the SDK produces
+ * @returns {SuggestionStream} the same answer in the domain's terms
+ */
+async function * toSuggestionStream<TOOLS extends ToolSet>(
+  parts: AsyncIterable<TextStreamPart<TOOLS>>
+): SuggestionStream {
+  for await (const part of parts) {
+    if (part.type === 'text-delta') {
+      yield {
+        type: 'text-delta',
+        delta: part.text,
+      };
+    }
+
+    if (part.type === 'error') {
+      yield {
+        type: 'error',
+        errorText: getErrorMessage(part.error),
+      };
+    }
+  }
+}
+
 /**
  * Interface for interacting with Vercel AI Gateway
  *
@@ -27,11 +66,24 @@ class VercelAIApi {
    */
   private readonly modelId: string;
 
+  /**
+   * Provider Gateway fallback order
+   */
+  private readonly providerOptions: ProviderOptions;
+
+  /**
+   * Set up model id and provider fallback order
+   */
   constructor() {
     /**
      * @todo make it dynamic, get from project settings
      */
     this.modelId = 'deepseek/deepseek-v4-flash';
+    this.providerOptions = {
+      gateway: {
+        order: ['novita', 'azure', 'deepseek'],
+      },
+    };
   }
 
   /**
@@ -45,15 +97,29 @@ class VercelAIApi {
       model: this.modelId,
       system,
       prompt,
-      providerOptions: {
-        gateway: {
-          order: ['novita', 'azure', 'deepseek'],
-        },
-      },
+      providerOptions: this.providerOptions,
     });
 
     return text;
   }
+
+  /**
+   * Send a system/prompt pair to the model and return the generated text as a stream
+   *
+   * @param {StreamParams} params - system instruction, prompt and abort signal
+   * @returns {SuggestionStream} text generated by the model, as it arrives
+   */
+  public stream({ system, prompt, signal }: StreamParams): SuggestionStream {
+    const { fullStream } = streamText({
+      model: this.modelId,
+      system,
+      prompt,
+      providerOptions: this.providerOptions,
+      abortSignal: signal,
+    });
+
+    return toSuggestionStream(fullStream);
+  }
 }
 
 export const vercelAIApi = new VercelAIApi();

src/services/askAi/index.ts

@@ -1 +1,2 @@
 export { AskAiService, askAiService } from './service';
+export { appendAiAssistantRoutes } from './routes';

src/services/askAi/routes.ts

@@ -0,0 +1,137 @@
+import '../../typeDefs/expressContext';
+import express from 'express';
+import { getEventsFactory } from '../../resolvers/helpers/eventsFactory';
+import { checkUserInWorkspaceByProjectId } from '../../directives/requireUserInWorkspace';
+import { askAiService } from './service';
+
+/**
+ * Verify the requesting user is a member of the project's workspace.
+ *
+ * @param req - Express request
+ * @param res - Express response
+ * @param projectId - project id from query parameters (may be string[] if repeated)
+ * @returns user id and validated project id if authorized, {@code null} otherwise (response already sent)
+ */
+async function authorizeProjectAccess(
+  req: express.Request,
+  res: express.Response,
+  projectId: unknown
+): Promise<{ userId: string; projectId: string } | null> {
+  const userId = req.context?.user?.id;
+
+  if (!userId) {
+    res.status(401).json({ error: 'Unauthorized. Please provide authorization token.' });
+
+    return null;
+  }
+
+  if (!projectId || typeof projectId !== 'string') {
+    res.status(400).json({ error: 'projectId query parameter is required' });
+
+    return null;
+  }
+
+  try {
+    await checkUserInWorkspaceByProjectId(req.context, projectId);
+  } catch (error) {
+    res.status(403).json({ error: error instanceof Error ? error.message : 'You have no access to this workspace' });
+
+    return null;
+  }
+
+  return {
+    userId,
+    projectId,
+  };
+}
+
+/**
+ * Create AI assistant router
+ *
+ * @returns Express router with AI assistant endpoints
+ */
+export function createAiStreamRouter(): express.Router {
+  const router = express.Router();
+
+  /**
+   * GET /integration/ai/stream?projectId=<projectId>&eventId=<eventId>&originalEventId=<originalEventId>
+   * Stream an AI suggestion for the event
+   */
+  router.get('/stream', async (req, res, next) => {
+    const abort = new AbortController();
+
+    /** Otherwise the model writes the rest of the answer, and bills for it, to nobody */
+    res.on('close', () => abort.abort());
+
+    try {
+      const { projectId, eventId, originalEventId } = req.query;
+
+      const authResult = await authorizeProjectAccess(req, res, projectId);
+
+      if (!authResult) {
+        return;
+      }
+
+      if (!eventId || typeof eventId !== 'string') {
+        res.status(400).json({ error: 'eventId query parameter is required' });
+
+        return;
+      }
+
+      if (!originalEventId || typeof originalEventId !== 'string') {
+        res.status(400).json({ error: 'originalEventId query parameter is required' });
+
+        return;
+      }
+
+      const eventsFactory = getEventsFactory(req.context, authResult.projectId);
+
+      let stream;
+
+      try {
+        stream = await askAiService.streamSuggestion(eventsFactory, eventId, originalEventId, abort.signal);
+      } catch (error) {
+        if (!(error instanceof Error) || error.message !== 'Event not found') {
+          throw error;
+        }
+
+        res.status(404).json({ error: error.message });
+
+        return;
+      }
+
+      res.writeHead(200, {
+        'content-type': 'text/event-stream',
+        'cache-control': 'no-cache',
+        connection: 'keep-alive',
+      });
+
+      for await (const part of stream) {
+        if (abort.signal.aborted) {
+          break;
+        }
+
+        res.write(`data: ${JSON.stringify(part)}\n\n`);
+      }
+
+      res.end();
+    } catch (error) {
+      if (abort.signal.aborted) {
+        return;
+      }
+
+      next(error);
+    }
+  });
+
+  return router;
+}
+
+/**
+ * Append AI assistant routes to Express app
+ *
+ * @param app - Express application instance
+ */
+export function appendAiAssistantRoutes(app: express.Application): void {
+  app.use('/integration/ai', createAiStreamRouter());
+}