feat(ai): stream AI suggestions over HTTP

package.json

@@ -1,6 +1,6 @@
 {
   "name": "hawk.api",
-  "version": "1.5.11",
+  "version": "1.5.12",
   "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,5 @@
-import { generateText } from 'ai';
+import { generateText, streamText } from 'ai';
+import { ProviderOptions } from '@ai-sdk/provider-utils';
 
 /**
  * Params for a single completion call to the model
@@ -29,11 +30,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'],
+      },
+    };
   }
 
   /**
@@ -47,15 +61,26 @@ 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 {CompletionParams} params - system instruction and prompt to complete
+   * @returns {StreamTextResult} text generated by the model, as a stream
+   */
+  public stream({ system, prompt }: CompletionParams): ReturnType<typeof streamText> {
+    return streamText({
+      model: this.modelId,
+      system,
+      prompt,
+      providerOptions: this.providerOptions,
+    });
+  }
 }
 
 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,114 @@
+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) => {
+    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 result;
+
+      try {
+        result = await askAiService.streamSuggestion(eventsFactory, eventId, originalEventId);
+      } catch (error) {
+        if (!(error instanceof Error) || error.message !== 'Event not found') {
+          throw error;
+        }
+
+        res.status(404).json({ error: error.message });
+
+        return;
+      }
+
+      result.pipeTextStreamToResponse(res);
+    } catch (error) {
+      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());
+}

src/services/askAi/service.ts

@@ -4,6 +4,7 @@ import { buildEventPrompt, spotlightInstruction } from './security/spotlighting'
 import { isLeaked, SUGGESTION_FALLBACK_MESSAGE } from './security/leakDetector';
 import { ctoInstruction } from './instructions/cto';
 import { EventsFactoryInterface } from '../types';
+import type { Event } from '../types';
 
 /**
  * Report that the leak tripwire fired.
@@ -43,12 +44,12 @@ export class AskAiService {
    * @param originalEventId - original event id
    * @returns {Promise<string>} - suggestion
    */
-  public async generateSuggestion(eventsFactory: EventsFactoryInterface, eventId: string, originalEventId: string): Promise<string> {
-    const event = await eventsFactory.getEventRepetition(eventId, originalEventId);
-
-    if (!event) {
-      throw new Error('Event not found');
-    }
+  public async generateSuggestion(
+    eventsFactory: EventsFactoryInterface,
+    eventId: string,
+    originalEventId: string
+  ): Promise<string> {
+    const event = await this.getEventOrThrow(eventsFactory, eventId, originalEventId);
 
     const { prompt, nonce } = buildEventPrompt(event.payload);
 
@@ -65,6 +66,61 @@ export class AskAiService {
 
     return text;
   }
+
+  /**
+   * Generate streaming suggestion for the event
+   *
+   * The payload is spotlighted by {@link buildEventPrompt} exactly as in
+   * {@link AskAiService.generateSuggestion}.
+   *
+   * @param eventsFactory - events factory
+   * @param eventId - event id
+   * @param originalEventId - original event id
+   * @returns streaming suggestion
+   */
+  public async streamSuggestion(
+    eventsFactory: EventsFactoryInterface,
+    eventId: string,
+    originalEventId: string
+  ): Promise<ReturnType<typeof vercelAIApi.stream>> {
+    const event = await this.getEventOrThrow(eventsFactory, eventId, originalEventId);
+
+    const { prompt, nonce } = buildEventPrompt(event.payload);
+
+    return vercelAIApi.stream({
+      system: ctoInstruction + spotlightInstruction(nonce),
+      prompt,
+    });
+  }
+
+  /**
+   * Find the event repetition or throw if it doesn't exist. A thrown lookup
+   * failure is normalized to the same message too, so it doesn't leak details.
+   *
+   * @param eventsFactory - events factory
+   * @param eventId - event id
+   * @param originalEventId - original event id
+   * @returns {Promise<Event>} - event repetition
+   */
+  private async getEventOrThrow(
+    eventsFactory: EventsFactoryInterface,
+    eventId: string,
+    originalEventId: string
+  ): Promise<Event> {
+    let event: Event | null;
+
+    try {
+      event = await eventsFactory.getEventRepetition(eventId, originalEventId);
+    } catch {
+      throw new Error('Event not found');
+    }
+
+    if (!event) {
+      throw new Error('Event not found');
+    }
+
+    return event;
+  }
 }
 
 export const askAiService = new AskAiService();

src/services/types.ts

@@ -3,7 +3,7 @@ import { EventAddons, EventData } from '@hawk.so/types';
 /**
  * Event type which is returned by events factory
  */
-type Event = {
+export type Event = {
     _id: string;
     payload: EventData<EventAddons>;
 };
@@ -20,4 +20,4 @@ export interface EventsFactoryInterface {
    * @returns {Promise<EventData<EventAddons>>} - event repetition
    */
   getEventRepetition(repetitionId: string, originalEventId: string): Promise<Event>;
-}
+}