From d691c1f5cf7e6494f5c892023c85c812c9a1f936 Mon Sep 17 00:00:00 2001 From: Julien Gautier Date: Tue, 19 May 2026 22:27:30 +0200 Subject: [PATCH] =?UTF-8?q?feat(portal-bff):=20ai-bridge=20controller=20?= =?UTF-8?q?=E2=80=94=20SSE=20chat=20+=20JSON=20rag/models?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Wires the AI relay into the BFF per ADR-0024 (now accepted): - AiBridgeController exposes POST /api/ai/chat (text/event-stream), GET /api/ai/rag/search, GET /api/ai/models under /api/ai/*. Session-gated (req.session.user) without an admin role — AI is an end-user surface. CSRF and double-submit token handled by the global middleware mounted in main.ts. - sse.writer translates ChatEvent oneof cases to kebab-case SSE event names (token / citation / agent-step / tool-call / error / done), JSON-encodes the inner payload, and produces a relay-side error frame on upstream failures (mapped from gRPC Status codes into the urn:apf-ai:* namespace the AI service already uses). - DTOs (class-validator) cap message count + length, restrict the role enum to user/assistant/system (tool messages are constructed caller-side by the BFF when tool dispatch lands), and bound RAG top_k between 1 and 50. - Browser disconnect (req 'close' event) → AbortController → call.cancel(), upstream LLM stops without leaking the stream. - AiClientModule now implements OnApplicationShutdown to close the four gRPC stubs on SIGTERM/SIGINT; main.ts calls app.enableShutdownHooks() to make it fire. - ADR-0024 promoted from proposed to accepted; README index + the CLAUDE.md roll-up updated. apps/portal-bff/.env.example documents AI_SERVICE_GRPC_ENDPOINT / CLIENT_ID / TLS for local boot. --- CLAUDE.md | 4 +- apps/portal-bff/.env.example | 13 + apps/portal-bff/src/app/app.module.ts | 2 + .../ai-bridge/ai-bridge.controller.spec.ts | 396 ++++++++++++++++++ .../grpc/ai-bridge/ai-bridge.controller.ts | 232 ++++++++++ .../src/grpc/ai-bridge/ai-bridge.module.ts | 24 ++ .../grpc/ai-bridge/dto/chat-request.dto.ts | 77 ++++ .../ai-bridge/dto/rag-search-query.dto.ts | 31 ++ .../src/grpc/ai-bridge/sse.writer.spec.ts | 81 ++++ .../src/grpc/ai-bridge/sse.writer.ts | 70 ++++ .../src/grpc/ai-client/ai-client.module.ts | 39 +- apps/portal-bff/src/main.ts | 10 + .../0024-ai-service-relay-grpc-sse-bridge.md | 2 +- docs/decisions/README.md | 2 +- 14 files changed, 971 insertions(+), 12 deletions(-) create mode 100644 apps/portal-bff/src/grpc/ai-bridge/ai-bridge.controller.spec.ts create mode 100644 apps/portal-bff/src/grpc/ai-bridge/ai-bridge.controller.ts create mode 100644 apps/portal-bff/src/grpc/ai-bridge/ai-bridge.module.ts create mode 100644 apps/portal-bff/src/grpc/ai-bridge/dto/chat-request.dto.ts create mode 100644 apps/portal-bff/src/grpc/ai-bridge/dto/rag-search-query.dto.ts create mode 100644 apps/portal-bff/src/grpc/ai-bridge/sse.writer.spec.ts create mode 100644 apps/portal-bff/src/grpc/ai-bridge/sse.writer.ts diff --git a/CLAUDE.md b/CLAUDE.md index 54c7d5c..7d47913 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -52,17 +52,19 @@ The structural, security, observability, and quality choices are recorded as ADR - **Local quality gates:** Husky + lint-staged + commitlint with Conventional Commits — see [ADR-0007](docs/decisions/0007-pre-commit-hooks-and-conventional-commits.md). - **Documentation site:** `docs/**/*.md` rendered as a separate static site via **VitePress** (Vite-based, Node-only toolchain, Markdown-first). Mermaid diagrams via `vitepress-plugin-mermaid`. Deployed on its own hostname behind the shared reverse-proxy; CI hook on `docs/` changes rebuilds + publishes. Decoupled from the apps — content lives in `docs/`, no in-app Markdown viewer — see [ADR-0022](docs/decisions/0022-docs-site-vitepress.md). - **Charts + dashboards:** `D3 + Observable Plot` wrapped in `libs/shared/charts/`, one Angular component per chart type (bar, donut, line, stacked-bar, …). A11y baked in by the lib (SVG ``/`<desc>`, `<details>` tabular fallback, colour-blind-safe palettes, AA-contrast text, `prefers-reduced-motion` gate). Bundle stays under [ADR-0017](docs/decisions/0017-performance-budgets-lighthouse-ci.md)'s lazy-chunk cap via per-`d3-*` module tree-shaking. Future bespoke visualisations land in raw D3 inside the same lib — see [ADR-0023](docs/decisions/0023-charts-d3-observable-plot.md). +- **AI service relay:** dedicated `apf-ai-service` repo (ASP.NET Core, Microsoft Agent Framework) consumed via native gRPC HTTP/2 only — proto contract vendored under `apps/portal-bff/src/grpc/proto/apf-ai/` with `ts-proto` codegen committed alongside. BFF dials with `@grpc/grpc-js` (h2c in dev, h2 + TLS in prod), bridges `ChatService.Chat` to `text/event-stream` for the SPA, exposes `RagService.Search` and `ModelsService.ListModels` as plain JSON endpoints. Identity travels as an unsigned `Principal` (subject, roles, attributes) in the proto body for the POC, hashed via the audit module's `HashUserIdService` so portal and AI service audit trails join on the same `actor_id_hash`. Production hardening (signed envelope vs mTLS) deferred — see [ADR-0024](docs/decisions/0024-ai-service-relay-grpc-sse-bridge.md). - **Runtime:** Node.js latest LTS major. ## Repository status The Nx workspace is **scaffolded and operational**. The three apps (`portal-shell`, `portal-admin`, `portal-bff`) and the four lib roots (`libs/feature/`, `libs/shared/state`, `libs/shared/tokens`, `libs/shared/ui`, `libs/shared/util`) are in place; CI runs `format:check / lint / test / build` on every PR. -ADRs 0001 → 0023 are accepted and cover the structural, security, observability, quality, i18n, admin-app, docs-site, and charts choices. **Shipped on `main`:** +ADRs 0001 → 0024 are accepted and cover the structural, security, observability, quality, i18n, admin-app, docs-site, charts, and AI-relay choices. **Shipped on `main`:** - **Phase-1 foundation** — Nx workspace, Angular `portal-shell`, NestJS `portal-bff`, Prisma + Postgres, Pino + OpenTelemetry, Husky/lint-staged/commitlint, Gitea Actions CI. - **Phase-2 auth + audit + security** — OIDC Auth Code + PKCE via MSAL Node, Redis sessions with AES-256-GCM at rest, idle 30 min sliding + absolute 12 h hard ceiling, RP-initiated logout, double-submit CSRF, `audit.events` append-only schema with role-based grants, helmet + env-driven CORS allowlist + rate limiting + structured error envelope (see [ADR-0021](docs/decisions/0021-phase-2-security-baseline.md)). - **Phase-3a admin app skeleton** — `portal-admin` SPA exists with brand tokens and routing; business modules (CMS, menu management, user list, audit log viewer) not yet implemented. +- **AI relay surface** — vendored protos + `AiClientModule` (gRPC clients, Principal mapper, metadata builder) + `AiBridgeController` exposing `POST /api/ai/chat` (SSE), `GET /api/ai/rag/search`, `GET /api/ai/models` (see [ADR-0024](docs/decisions/0024-ai-service-relay-grpc-sse-bridge.md)). Live consumer (chatbot widget on `portal-shell`) and the proto-drift CI gate ship next. **Still on the roadmap:** diff --git a/apps/portal-bff/.env.example b/apps/portal-bff/.env.example index b51761b..1b85a54 100644 --- a/apps/portal-bff/.env.example +++ b/apps/portal-bff/.env.example @@ -206,3 +206,16 @@ CORS_ALLOWED_ORIGINS=http://localhost:4200,http://localhost:4300 # BFF_JWKS_KID — wired # <SERVICE>_API_BASE_URL (per integrated downstream — lands with the first integration) # <SERVICE>_TIMEOUT_MS (optional, defaults to 5000 — lands with the first integration) + +# AI service relay (ADR-0024) — the BFF dials apf-ai-service over +# native gRPC HTTP/2 and bridges chat streams to SSE for the SPA. +# apf-ai-service runs from its own repo (../apf-ai-service); use +# that repo's docker-compose.yml to bring it up locally, then point +# AI_SERVICE_GRPC_ENDPOINT here at the host-published port. No +# JWT/auth on the wire in v1 — the Principal travels in the proto +# body (per ADR-0024 §"Sub-decision 4 — POC unsigned principal"). +AI_SERVICE_GRPC_ENDPOINT=localhost:8080 +AI_SERVICE_CLIENT_ID=apf-portal-dev +# Set to 'true' in preprod/prod (h2 + TLS via the edge proxy); +# 'false' in dev for h2c against the local apf-ai-service. +AI_SERVICE_GRPC_TLS=false diff --git a/apps/portal-bff/src/app/app.module.ts b/apps/portal-bff/src/app/app.module.ts index ae5e43c..1d1e394 100644 --- a/apps/portal-bff/src/app/app.module.ts +++ b/apps/portal-bff/src/app/app.module.ts @@ -8,6 +8,7 @@ import { AdminModule } from '../admin/admin.module'; import { AuditModule } from '../audit/audit.module'; import { AuthModule } from '../auth/auth.module'; import { DownstreamModule } from '../downstream/downstream.module'; +import { AiBridgeModule } from '../grpc/ai-bridge/ai-bridge.module'; import { MeModule } from '../me/me.module'; import { RedisModule } from '../redis/redis.module'; import { SecurityModule } from '../security/security.module'; @@ -25,6 +26,7 @@ import { UsersModule } from '../users/users.module'; SecurityModule, HealthModule, AdminModule, + AiBridgeModule, DownstreamModule, MeModule, UsersModule, diff --git a/apps/portal-bff/src/grpc/ai-bridge/ai-bridge.controller.spec.ts b/apps/portal-bff/src/grpc/ai-bridge/ai-bridge.controller.spec.ts new file mode 100644 index 0000000..bb324fc --- /dev/null +++ b/apps/portal-bff/src/grpc/ai-bridge/ai-bridge.controller.spec.ts @@ -0,0 +1,396 @@ +import { EventEmitter } from 'node:events'; +import { randomBytes } from 'node:crypto'; +import { describe, it, expect, beforeAll, afterAll, beforeEach } from '@jest/globals'; +import { UnauthorizedException } from '@nestjs/common'; +import { + ChannelCredentials, + Server, + ServerCredentials, + status as GrpcStatus, + type sendUnaryData, + type ServerUnaryCall, + type ServerWritableStream, +} from '@grpc/grpc-js'; +import { HashUserIdService } from '../../audit/hash-user-id.service'; +import { ChatClient } from '../ai-client/chat.client'; +import { GrpcMetadataBuilder } from '../ai-client/grpc-metadata.builder'; +import { ModelsClient } from '../ai-client/models.client'; +import { PrincipalMapper } from '../ai-client/principal.mapper'; +import { RagClient } from '../ai-client/rag.client'; +import { + ChatServiceClient, + ChatServiceService, + type ChatEvent, + type ChatRequest, +} from '../gen/apf-ai/chat'; +import { + ModelsServiceClient, + ModelsServiceService, + type ListModelsRequest, + type ListModelsResponse, +} from '../gen/apf-ai/models'; +import { + RagServiceClient, + RagServiceService, + type RagSearchRequest, + type RagSearchResponse, +} from '../gen/apf-ai/rag'; +import { Struct } from '../gen/apf-ai/google/protobuf/struct'; +import { AiBridgeController } from './ai-bridge.controller'; +import type { AiServiceConfig } from '../../config/check-ai-service-config'; +import type { AuthenticatedUser } from '../../auth/auth.service'; + +/** + * Exercises `AiBridgeController` end-to-end through the gRPC wire + * against an in-process fake of every service the controller calls. + * + * The controller is a thin translation layer (DTO → proto, gRPC → + * SSE, session → Principal). The spec confirms each direction of + * the translation lands the right bytes. + */ + +const STRONG_SALT = randomBytes(32).toString('base64url'); +const ORIGINAL_SALT = process.env['LOG_USER_ID_SALT']; + +beforeAll(() => { + process.env['LOG_USER_ID_SALT'] = STRONG_SALT; +}); + +afterAll(() => { + if (ORIGINAL_SALT === undefined) { + delete process.env['LOG_USER_ID_SALT']; + } else { + process.env['LOG_USER_ID_SALT'] = ORIGINAL_SALT; + } +}); + +const USER: AuthenticatedUser = { + oid: 'user-oid-42', + tid: 'tenant-1', + username: 'alice@example.org', + displayName: 'Alice', + amr: ['pwd', 'mfa'], + roles: ['admin'], +}; + +// ---- Fake gRPC server (Chat + Rag + Models) ------------------- + +let server: Server; +let port: number; +let chatHandler: (call: ServerWritableStream<ChatRequest, ChatEvent>) => void; +let ragHandler: ( + call: ServerUnaryCall<RagSearchRequest, RagSearchResponse>, + callback: sendUnaryData<RagSearchResponse>, +) => void; +let modelsHandler: ( + call: ServerUnaryCall<ListModelsRequest, ListModelsResponse>, + callback: sendUnaryData<ListModelsResponse>, +) => void; +let observedChatRequest: ChatRequest | null = null; + +beforeAll(async () => { + server = new Server(); + server.addService(ChatServiceService, { + chat: (call: ServerWritableStream<ChatRequest, ChatEvent>) => { + observedChatRequest = call.request; + chatHandler(call); + }, + }); + server.addService(RagServiceService, { + search: ( + call: ServerUnaryCall<RagSearchRequest, RagSearchResponse>, + callback: sendUnaryData<RagSearchResponse>, + ) => ragHandler(call, callback), + }); + server.addService(ModelsServiceService, { + listModels: ( + call: ServerUnaryCall<ListModelsRequest, ListModelsResponse>, + callback: sendUnaryData<ListModelsResponse>, + ) => modelsHandler(call, callback), + }); + + port = await new Promise<number>((resolve, reject) => { + server.bindAsync('127.0.0.1:0', ServerCredentials.createInsecure(), (err, bound) => { + if (err) { + reject(err); + return; + } + resolve(bound); + }); + }); +}); + +afterAll(async () => { + await new Promise<void>((resolve, reject) => { + server.tryShutdown((err) => { + if (err) { + reject(err); + return; + } + resolve(); + }); + }); +}); + +// ---- Controller wiring -------------------------------------- + +function buildController(): AiBridgeController { + const config: AiServiceConfig = { + endpoint: `127.0.0.1:${port}`, + clientId: 'apf-portal-test', + useTls: false, + }; + const credentials = ChannelCredentials.createInsecure(); + const metadata = new GrpcMetadataBuilder(config); + const chatClient = new ChatClient(new ChatServiceClient(config.endpoint, credentials), metadata); + const ragClient = new RagClient(new RagServiceClient(config.endpoint, credentials), metadata); + const modelsClient = new ModelsClient( + new ModelsServiceClient(config.endpoint, credentials), + metadata, + ); + const principalMapper = new PrincipalMapper(new HashUserIdService()); + return new AiBridgeController(chatClient, ragClient, modelsClient, principalMapper); +} + +// ---- Mock Express Request / Response ----------------------- + +interface CapturedResponse { + statusCode: number | undefined; + headers: Record<string, string>; + body: string; + ended: boolean; + writableEnded: boolean; +} + +function mockReq(user: AuthenticatedUser | undefined): { + req: import('express').Request; + close: () => void; +} { + const emitter = new EventEmitter(); + const req = Object.assign(emitter, { + session: { user }, + }) as unknown as import('express').Request; + return { + req, + close: () => emitter.emit('close'), + }; +} + +function mockRes(): { res: import('express').Response; captured: CapturedResponse } { + const captured: CapturedResponse = { + statusCode: undefined, + headers: {}, + body: '', + ended: false, + writableEnded: false, + }; + const res = { + set statusCode(value: number) { + captured.statusCode = value; + }, + get writableEnded() { + return captured.writableEnded; + }, + setHeader(name: string, value: string): void { + captured.headers[name] = value; + }, + flushHeaders(): void { + // no-op for the mock + }, + write(chunk: string | Buffer): boolean { + captured.body += typeof chunk === 'string' ? chunk : chunk.toString(); + return true; + }, + end(chunk?: string | Buffer): void { + if (chunk !== undefined) { + captured.body += typeof chunk === 'string' ? chunk : chunk.toString(); + } + captured.ended = true; + captured.writableEnded = true; + }, + }; + return { res: res as unknown as import('express').Response, captured }; +} + +beforeEach(() => { + observedChatRequest = null; +}); + +// --------------------------------------------------------------- + +describe('AiBridgeController — POST /api/ai/chat (SSE bridge)', () => { + it('streams ChatEvent frames as SSE and sets the right headers', async () => { + chatHandler = (call) => { + call.write({ token: { token: 'hi', value: 'Hi' } }); + call.write({ token: { token: ' there', value: ' there' } }); + call.write({ done: { stats: { tokensIn: 1, tokensOut: 2, chunksRetrieved: 0 } } }); + call.end(); + }; + + const controller = buildController(); + const { req } = mockReq(USER); + const { res, captured } = mockRes(); + + await controller.chat( + { + messages: [{ role: 'user', content: 'hello' }], + conversationId: 'c-1', + }, + req, + res, + ); + + expect(captured.headers['Content-Type']).toBe('text/event-stream; charset=utf-8'); + expect(captured.headers['Cache-Control']).toBe('no-cache, no-transform'); + expect(captured.headers['X-Accel-Buffering']).toBe('no'); + expect(captured.body).toContain('event: token\n'); + expect(captured.body).toContain('"value":"Hi"'); + expect(captured.body).toContain('event: done\n'); + expect(captured.ended).toBe(true); + }); + + it('places a hashed subject and pass-through roles in the proto Principal', async () => { + chatHandler = (call) => { + call.write({ done: { stats: { tokensIn: 0, tokensOut: 0, chunksRetrieved: 0 } } }); + call.end(); + }; + const controller = buildController(); + const { req } = mockReq(USER); + const { res } = mockRes(); + + await controller.chat( + { + messages: [{ role: 'user', content: 'q' }], + }, + req, + res, + ); + + expect(observedChatRequest?.principal?.subject).toMatch(/^[0-9a-f]{16}$/); + expect(observedChatRequest?.principal?.subject).not.toBe(USER.oid); + expect(observedChatRequest?.principal?.roles).toEqual(['admin']); + expect(observedChatRequest?.principal?.attributes).toEqual({ tenantId: 'tenant-1' }); + expect(observedChatRequest?.toolsAvailable).toEqual([]); + }); + + it('rejects with 401 when no session.user is present', async () => { + const controller = buildController(); + const { req } = mockReq(undefined); + const { res } = mockRes(); + + await expect( + controller.chat({ messages: [{ role: 'user', content: 'q' }] }, req, res), + ).rejects.toBeInstanceOf(UnauthorizedException); + }); + + it('cancels the gRPC call when the request emits "close" mid-stream', async () => { + const cancellationObserved = new Promise<void>((resolve) => { + chatHandler = (call) => { + call.write({ token: { token: 't', value: 't' } }); + call.on('cancelled', () => resolve()); + // never end — wait for client cancellation + }; + }); + + const controller = buildController(); + const { req, close } = mockReq(USER); + const { res } = mockRes(); + + const completion = controller.chat({ messages: [{ role: 'user', content: 'q' }] }, req, res); + + // Give the server time to emit the first frame then cut the + // request — this is the controller's documented browser-close + // pathway. + setTimeout(close, 30); + + await completion; + await cancellationObserved; + }); + + it('writes a relay error frame on upstream failure', async () => { + chatHandler = (call) => { + call.emit('error', { + code: GrpcStatus.UNAVAILABLE, + details: 'upstream down', + message: 'upstream down', + }); + }; + + const controller = buildController(); + const { req } = mockReq(USER); + const { res, captured } = mockRes(); + + await controller.chat({ messages: [{ role: 'user', content: 'q' }] }, req, res); + + expect(captured.body).toContain('event: error\n'); + expect(captured.body).toContain('"code":"urn:apf-ai:unavailable"'); + expect(captured.body).toContain('"retriable":true'); + expect(captured.ended).toBe(true); + }); +}); + +describe('AiBridgeController — GET /api/ai/rag/search', () => { + it('returns the unary response', async () => { + ragHandler = (_call, callback) => { + callback(null, { + chunks: [ + { + id: 'k', + documentId: 'd', + content: 'snippet', + source: 'src', + score: 1, + metadata: Struct.fromPartial({}), + }, + ], + correlationId: 'corr-1', + }); + }; + + const controller = buildController(); + const { req } = mockReq(USER); + const response = await controller.ragSearch({ query: 'hello', topK: 3 }, req); + + expect(response.chunks).toHaveLength(1); + expect(response.correlationId).toBe('corr-1'); + }); + + it('rejects with 401 when no session.user', async () => { + const controller = buildController(); + const { req } = mockReq(undefined); + await expect(controller.ragSearch({ query: 'x' }, req)).rejects.toBeInstanceOf( + UnauthorizedException, + ); + }); +}); + +describe('AiBridgeController — GET /api/ai/models', () => { + it('returns the model list', async () => { + modelsHandler = (_call, callback) => { + callback(null, { + active: 'openai-compatible', + providers: [ + { + discriminator: 'openai-compatible', + capabilities: 'chat,embedding', + endpoint: 'http://ollama:11434/v1', + model: 'qwen2.5:3b', + embeddingModel: 'nomic-embed-text', + }, + ], + }); + }; + + const controller = buildController(); + const { req } = mockReq(USER); + const response = await controller.listModels(req); + expect(response.active).toBe('openai-compatible'); + expect(response.providers).toHaveLength(1); + }); + + it('rejects with 401 when no session.user', async () => { + const controller = buildController(); + const { req } = mockReq(undefined); + await expect(controller.listModels(req)).rejects.toBeInstanceOf(UnauthorizedException); + }); +}); diff --git a/apps/portal-bff/src/grpc/ai-bridge/ai-bridge.controller.ts b/apps/portal-bff/src/grpc/ai-bridge/ai-bridge.controller.ts new file mode 100644 index 0000000..6d787df --- /dev/null +++ b/apps/portal-bff/src/grpc/ai-bridge/ai-bridge.controller.ts @@ -0,0 +1,232 @@ +import { + Body, + Controller, + Get, + HttpStatus, + Post, + Query, + Req, + Res, + UnauthorizedException, +} from '@nestjs/common'; +import { ApiCookieAuth, ApiOperation, ApiTags } from '@nestjs/swagger'; +import { status as GrpcStatus, type ServiceError } from '@grpc/grpc-js'; +import type { Request, Response } from 'express'; +import { ChatClient } from '../ai-client/chat.client'; +import { ModelsClient } from '../ai-client/models.client'; +import { PrincipalMapper } from '../ai-client/principal.mapper'; +import { RagClient } from '../ai-client/rag.client'; +import type { ChatEvent, ChatRequest } from '../gen/apf-ai/chat'; +import { ChatRole } from '../gen/apf-ai/common'; +import type { ListModelsResponse } from '../gen/apf-ai/models'; +import type { RagSearchResponse } from '../gen/apf-ai/rag'; +import { ChatRequestDto, type ChatMessageDto } from './dto/chat-request.dto'; +import { RagSearchQueryDto } from './dto/rag-search-query.dto'; +import { chatEventToSseFrame, relayErrorFrame } from './sse.writer'; +import type { AuthenticatedUser } from '../../auth/auth.service'; + +/** + * BFF-facing surface of the AI relay, per + * [ADR-0024](../../../../../docs/decisions/0024-ai-service-relay-grpc-sse-bridge.md): + * + * - `POST /api/ai/chat` — streaming chat, bridged from + * `ChatService.Chat` (gRPC server- + * stream) to `text/event-stream`. + * - `GET /api/ai/rag/search` — unary RAG retrieval. + * - `GET /api/ai/models` — list configured providers. + * + * No `@RequireAdmin()` — AI features are end-user surfaces, gated + * only by the active portal session (`req.session.user`). The + * session + CSRF middleware mounted in `main.ts` covers the + * authentication + double-submit token before the controller + * runs; this class asserts presence of `session.user` for the + * routes that need it and lets the global middleware handle the + * rest. + */ +@ApiTags('ai') +@ApiCookieAuth('portal_session') +@Controller('ai') +export class AiBridgeController { + constructor( + private readonly chatClient: ChatClient, + private readonly ragClient: RagClient, + private readonly modelsClient: ModelsClient, + private readonly principalMapper: PrincipalMapper, + ) {} + + @ApiOperation({ + summary: + 'Streaming chat. Returns text/event-stream with one frame per ChatEvent (token / citation / agent-step / tool-call / error / done).', + }) + @Post('chat') + async chat( + @Body() body: ChatRequestDto, + @Req() req: Request, + @Res() res: Response, + ): Promise<void> { + const user = requireSessionUser(req); + + const principal = this.principalMapper.fromInputs({ + oid: user.oid, + tid: user.tid, + roles: user.roles, + }); + const protoRequest: ChatRequest = { + messages: body.messages.map(toProtoMessage), + conversationId: body.conversationId ?? '', + model: body.model ?? '', + provider: body.provider ?? '', + toolsAvailable: [], + // v1: tool registry is empty (per ADR-0024 §"Tool-dispatch + // contract"). The AI service will never emit `tool_call` for + // requests with `toolsAvailable: []`; the SSE writer still + // covers the case in case a future tool lands. + rag: { enabled: false, topK: 0 }, + principal, + }; + + writeSseHeaders(res); + + // Browser disconnect (tab close, fetch abort, network drop) → + // AbortController → gRPC call.cancel() → upstream LLM stop. + // Per ADR-0024 §"SSE bridge between BFF and SPA". + const abort = new AbortController(); + req.on('close', () => { + if (!res.writableEnded) { + abort.abort(); + } + }); + + const stream = this.chatClient.chat(protoRequest, { signal: abort.signal }); + + try { + for await (const event of stream as AsyncIterable<ChatEvent>) { + const frame = chatEventToSseFrame(event); + if (frame !== null) { + res.write(frame); + } + } + } catch (err) { + // gRPC `CANCELLED` after `abort.abort()` is the expected exit + // path on browser disconnect — do not surface an error frame + // because the response is already closing. Anything else + // becomes a structured error event so the SPA's renderer + // sees the failure rather than a torn-down connection. + const code = (err as Partial<ServiceError>).code; + const cancelled = code === GrpcStatus.CANCELLED; + if (!cancelled && !res.writableEnded) { + res.write( + relayErrorFrame( + mapServiceErrorCode(code), + (err as Error).message ?? 'AI relay error', + isRetriable(code), + ), + ); + } + } finally { + if (!res.writableEnded) { + res.end(); + } + } + } + + @ApiOperation({ summary: 'Unary RAG retrieval bounded by the caller principal.' }) + @Get('rag/search') + async ragSearch( + @Query() query: RagSearchQueryDto, + @Req() req: Request, + ): Promise<RagSearchResponse> { + const user = requireSessionUser(req); + const principal = this.principalMapper.fromInputs({ + oid: user.oid, + tid: user.tid, + roles: user.roles, + }); + + return this.ragClient.search({ + query: query.query, + topK: query.topK ?? 0, + filters: { + source: query.source ?? '', + documentId: query.documentId ?? '', + }, + principal, + }); + } + + @ApiOperation({ summary: 'List configured AI providers and the active provider.' }) + @Get('models') + async listModels(@Req() req: Request): Promise<ListModelsResponse> { + requireSessionUser(req); + return this.modelsClient.listModels({}); + } +} + +function requireSessionUser(req: Request): AuthenticatedUser { + const user = req.session.user; + if (!user) { + throw new UnauthorizedException({ + code: 'unauthenticated', + message: 'The AI surface requires an authenticated portal session.', + }); + } + return user; +} + +function writeSseHeaders(res: Response): void { + res.statusCode = HttpStatus.OK; + res.setHeader('Content-Type', 'text/event-stream; charset=utf-8'); + res.setHeader('Cache-Control', 'no-cache, no-transform'); + res.setHeader('Connection', 'keep-alive'); + // nginx-style buffering hint — keeps reverse-proxies from + // accumulating chunks before forwarding, which would defeat the + // streaming UX. + res.setHeader('X-Accel-Buffering', 'no'); + res.flushHeaders(); +} + +function toProtoMessage(message: ChatMessageDto): { + role: ChatRole; + content: string; + toolCallId: string; + name: string; +} { + return { + role: ROLE_PROTO[message.role], + content: message.content, + toolCallId: '', + name: '', + }; +} + +const ROLE_PROTO: Record<ChatMessageDto['role'], ChatRole> = { + system: ChatRole.CHAT_ROLE_SYSTEM, + user: ChatRole.CHAT_ROLE_USER, + assistant: ChatRole.CHAT_ROLE_ASSISTANT, +}; + +/** + * gRPC status → SSE error code. Keeps the urn:apf-ai:* namespace + * the AI service itself uses so a relay-side error and an upstream + * error look the same to the SPA's renderer. + */ +function mapServiceErrorCode(code: number | undefined): string { + switch (code) { + case GrpcStatus.UNAVAILABLE: + return 'urn:apf-ai:unavailable'; + case GrpcStatus.DEADLINE_EXCEEDED: + return 'urn:apf-ai:timeout'; + case GrpcStatus.PERMISSION_DENIED: + return 'urn:apf-ai:permission_denied'; + case GrpcStatus.RESOURCE_EXHAUSTED: + return 'urn:apf-ai:rate_limited'; + case GrpcStatus.INVALID_ARGUMENT: + return 'urn:apf-ai:invalid_argument'; + default: + return 'urn:apf-ai:relay_error'; + } +} + +function isRetriable(code: number | undefined): boolean { + return code === GrpcStatus.UNAVAILABLE || code === GrpcStatus.DEADLINE_EXCEEDED; +} diff --git a/apps/portal-bff/src/grpc/ai-bridge/ai-bridge.module.ts b/apps/portal-bff/src/grpc/ai-bridge/ai-bridge.module.ts new file mode 100644 index 0000000..273f129 --- /dev/null +++ b/apps/portal-bff/src/grpc/ai-bridge/ai-bridge.module.ts @@ -0,0 +1,24 @@ +import { Module } from '@nestjs/common'; +import { AiClientModule } from '../ai-client/ai-client.module'; +import { AiBridgeController } from './ai-bridge.controller'; + +/** + * Hosts `AiBridgeController` and depends on `AiClientModule` for + * the four wrapper clients (`ChatClient`, `RagClient`, + * `ModelsClient`, plus `PrincipalMapper`). `AiClientModule`'s + * `OnApplicationShutdown` lifecycle closes the gRPC stubs at + * process termination — wiring this module into `AppModule` + * brings both the controller AND the shutdown contract along. + * + * Per [ADR-0024](../../../../../docs/decisions/0024-ai-service-relay-grpc-sse-bridge.md) + * §"Out of scope for this ADR" the ingestion surface is not + * exposed via the BFF in v1 — `IngestionClient` is in + * `AiClientModule` for future reuse but `AiBridgeController` + * does not surface it. The CLI under `apf-ai-service/tools/Apf.Ai.Ingest/` + * is the v1 ingestion path. + */ +@Module({ + imports: [AiClientModule], + controllers: [AiBridgeController], +}) +export class AiBridgeModule {} diff --git a/apps/portal-bff/src/grpc/ai-bridge/dto/chat-request.dto.ts b/apps/portal-bff/src/grpc/ai-bridge/dto/chat-request.dto.ts new file mode 100644 index 0000000..729a5d0 --- /dev/null +++ b/apps/portal-bff/src/grpc/ai-bridge/dto/chat-request.dto.ts @@ -0,0 +1,77 @@ +import { Type } from 'class-transformer'; +import { + ArrayMaxSize, + ArrayMinSize, + IsArray, + IsIn, + IsOptional, + IsString, + MaxLength, + MinLength, + ValidateNested, +} from 'class-validator'; + +/** + * Roles the SPA may attach to a message in the chat history. + * + * `tool` is intentionally absent — tool-result messages are + * constructed by the BFF itself (caller-side tool dispatch, per + * [ADR-0024](../../../../../../docs/decisions/0024-ai-service-relay-grpc-sse-bridge.md) + * §"Tool-dispatch contract — caller-side execution") and never + * provided by the SPA. v1 ships with an empty tool registry, so + * no `tool` message ever reaches this controller. + */ +const ALLOWED_ROLES = ['user', 'assistant', 'system'] as const; +type AllowedRole = (typeof ALLOWED_ROLES)[number]; + +export class ChatMessageDto { + @IsIn(ALLOWED_ROLES) + role!: AllowedRole; + + /** + * Free-form message body. The cap is generous (16 KB) to support + * code blocks and pasted excerpts; the BFF leaves further + * pre-processing (truncation, token counting) to the AI service. + */ + @IsString() + @MinLength(0) + @MaxLength(16_384) + content!: string; +} + +/** + * POST /api/ai/chat body. + * + * Stateless from the BFF's perspective — the SPA owns the + * conversation history. `conversationId` is an opaque correlation + * key used by the AI service's audit log (not a database key on + * either side); the SPA may omit it for one-off calls. + */ +export class ChatRequestDto { + @IsArray() + @ArrayMinSize(1) + @ArrayMaxSize(64) + @ValidateNested({ each: true }) + @Type(() => ChatMessageDto) + messages!: ChatMessageDto[]; + + @IsOptional() + @IsString() + @MaxLength(128) + conversationId?: string; + + /** + * Optional model + provider hints. Both default to "" on the + * wire, which the AI service interprets as "use the configured + * default". The SPA does not need to pass these in v1. + */ + @IsOptional() + @IsString() + @MaxLength(64) + model?: string; + + @IsOptional() + @IsString() + @MaxLength(64) + provider?: string; +} diff --git a/apps/portal-bff/src/grpc/ai-bridge/dto/rag-search-query.dto.ts b/apps/portal-bff/src/grpc/ai-bridge/dto/rag-search-query.dto.ts new file mode 100644 index 0000000..5e909d9 --- /dev/null +++ b/apps/portal-bff/src/grpc/ai-bridge/dto/rag-search-query.dto.ts @@ -0,0 +1,31 @@ +import { Type } from 'class-transformer'; +import { IsInt, IsOptional, IsString, Max, Min, MinLength } from 'class-validator'; + +/** + * Query-string DTO for `GET /api/ai/rag/search`. + * + * Bounded — the upstream RAG service has its own server-side cap on + * `top_k`, but the BFF rejects obviously-out-of-range values early + * so a single SPA bug cannot induce repeated 400s round-tripped + * through the AI service. + */ +export class RagSearchQueryDto { + @IsString() + @MinLength(1) + query!: string; + + @IsOptional() + @Type(() => Number) + @IsInt() + @Min(1) + @Max(50) + topK?: number; + + @IsOptional() + @IsString() + source?: string; + + @IsOptional() + @IsString() + documentId?: string; +} diff --git a/apps/portal-bff/src/grpc/ai-bridge/sse.writer.spec.ts b/apps/portal-bff/src/grpc/ai-bridge/sse.writer.spec.ts new file mode 100644 index 0000000..84d5d72 --- /dev/null +++ b/apps/portal-bff/src/grpc/ai-bridge/sse.writer.spec.ts @@ -0,0 +1,81 @@ +import { describe, it, expect } from '@jest/globals'; +import { chatEventToSseFrame, relayErrorFrame } from './sse.writer'; + +/** + * Locks the wire shape the SPA consumes. Each frame is: + * + * event: <name>\n + * data: <json>\n + * \n + * + * The terminal blank line is the SSE message separator. + */ + +describe('chatEventToSseFrame', () => { + it('maps token events with the JSON-encoded inner value', () => { + const frame = chatEventToSseFrame({ token: { token: 't1', value: 'hello' } }); + expect(frame).toBe(`event: token\ndata: {"token":"t1","value":"hello"}\n\n`); + }); + + it('maps citation events', () => { + const frame = chatEventToSseFrame({ + citation: { + chunkId: 'c-1', + documentId: 'd-1', + source: 's', + score: 0.42, + snippet: 'snip', + }, + }); + expect(frame?.startsWith('event: citation\n')).toBe(true); + expect(frame).toContain('"chunkId":"c-1"'); + }); + + it('maps agent_step to kebab-case `agent-step`', () => { + const frame = chatEventToSseFrame({ + agentStep: { agent: 'a', step: 's', stepId: '1' }, + }); + expect(frame?.startsWith('event: agent-step\n')).toBe(true); + }); + + it('maps tool_call to kebab-case `tool-call`', () => { + const frame = chatEventToSseFrame({ + toolCall: { callId: 'c-1', name: 'echo', args: undefined }, + }); + expect(frame?.startsWith('event: tool-call\n')).toBe(true); + }); + + it('maps error events', () => { + const frame = chatEventToSseFrame({ + error: { code: 'urn:apf-ai:foo', message: 'bad', retriable: false }, + }); + expect(frame).toBe( + `event: error\ndata: {"code":"urn:apf-ai:foo","message":"bad","retriable":false}\n\n`, + ); + }); + + it('maps done as the terminal frame', () => { + const frame = chatEventToSseFrame({ + done: { stats: { tokensIn: 1, tokensOut: 2, chunksRetrieved: 0 } }, + }); + expect(frame?.startsWith('event: done\n')).toBe(true); + expect(frame?.endsWith('\n\n')).toBe(true); + }); + + it('returns null when no oneof case is populated', () => { + expect(chatEventToSseFrame({})).toBeNull(); + }); +}); + +describe('relayErrorFrame', () => { + it('emits an error frame with the supplied code + message + retriable flag', () => { + expect(relayErrorFrame('urn:apf-ai:relay_error', 'kaboom', true)).toBe( + `event: error\ndata: {"code":"urn:apf-ai:relay_error","message":"kaboom","retriable":true}\n\n`, + ); + }); + + it('defaults `retriable` to false when omitted', () => { + const frame = relayErrorFrame('urn:apf-ai:foo', 'msg'); + expect(frame).toContain('"retriable":false'); + }); +}); diff --git a/apps/portal-bff/src/grpc/ai-bridge/sse.writer.ts b/apps/portal-bff/src/grpc/ai-bridge/sse.writer.ts new file mode 100644 index 0000000..da9588a --- /dev/null +++ b/apps/portal-bff/src/grpc/ai-bridge/sse.writer.ts @@ -0,0 +1,70 @@ +import type { ChatEvent } from '../gen/apf-ai/chat'; + +/** + * Translate one `apf.ai.v1.ChatEvent` into a single SSE frame for + * the SPA, per ADR-0024 §"Sub-decision 2 — SSE bridge between BFF + * and SPA". The mapping is intentionally one-to-one with the + * proto `oneof` cases: + * + * token → `event: token` / data = TokenEvent JSON + * citation → `event: citation` / data = CitationEvent JSON + * agent_step → `event: agent-step` / data = AgentStepEvent JSON + * tool_call → `event: tool-call` / data = ToolCallEvent JSON + * error → `event: error` / data = ErrorEvent JSON + * done → `event: done` / data = DoneEvent JSON + * + * SSE event names use kebab-case so the SPA's `EventSource` / + * fetch-streaming consumer can dispatch with `.addEventListener('agent-step', …)` + * without re-mapping proto camelCase. The terminal `done` frame is + * the contract's stream-close marker — no `[DONE]` sentinel, per + * ADR-0024. + * + * Returns `null` when the event carries no populated oneof case + * (defensive — gRPC-js will not produce this in practice, but the + * caller can safely skip on `null` rather than emit an empty + * frame). The data payload is JSON.stringified; consumers parse + * with `JSON.parse(event.data)`. + */ +export function chatEventToSseFrame(event: ChatEvent): string | null { + if (event.token !== undefined) { + return formatFrame('token', event.token); + } + if (event.citation !== undefined) { + return formatFrame('citation', event.citation); + } + if (event.agentStep !== undefined) { + return formatFrame('agent-step', event.agentStep); + } + if (event.toolCall !== undefined) { + return formatFrame('tool-call', event.toolCall); + } + if (event.error !== undefined) { + return formatFrame('error', event.error); + } + if (event.done !== undefined) { + return formatFrame('done', event.done); + } + return null; +} + +/** + * Convenience used by the controller's error path: synthesise a + * `urn:apf-ai:relay_error` event frame so the SPA receives a + * structured failure rather than a torn-down connection. Matches the + * shape of the AI service's own `ErrorEvent` so the SPA's renderer + * does not need a second code path for relay-level failures vs + * upstream model errors. + */ +export function relayErrorFrame(code: string, message: string, retriable = false): string { + return formatFrame('error', { code, message, retriable }); +} + +function formatFrame(eventName: string, data: unknown): string { + // SSE spec: every field line ends with `\n`, the frame is + // terminated by a blank line (`\n\n`). `data:` is followed by a + // single space convention. `JSON.stringify` produces no newlines + // for plain objects, so a single `data:` line is correct; a + // multi-line payload (not used by this codec) would require + // splitting on `\n` and prefixing each part with `data:`. + return `event: ${eventName}\ndata: ${JSON.stringify(data)}\n\n`; +} diff --git a/apps/portal-bff/src/grpc/ai-client/ai-client.module.ts b/apps/portal-bff/src/grpc/ai-client/ai-client.module.ts index 3f50292..7195330 100644 --- a/apps/portal-bff/src/grpc/ai-client/ai-client.module.ts +++ b/apps/portal-bff/src/grpc/ai-client/ai-client.module.ts @@ -1,5 +1,5 @@ -import { Module, type Provider } from '@nestjs/common'; -import { ChannelCredentials } from '@grpc/grpc-js'; +import { Inject, Module, type OnApplicationShutdown, type Provider } from '@nestjs/common'; +import { ChannelCredentials, type Client } from '@grpc/grpc-js'; import { HashUserIdService } from '../../audit/hash-user-id.service'; import { assertAiServiceConfig, type AiServiceConfig } from '../../config/check-ai-service-config'; import { ChatServiceClient } from '../gen/apf-ai/chat'; @@ -122,11 +122,32 @@ const grpcStubProviders: Provider[] = [ ], exports: [ChatClient, RagClient, IngestionClient, ModelsClient, PrincipalMapper], }) -export class AiClientModule { - // Lifecycle (channel close on shutdown) lands when the SSE-bridge - // PR wires this module into AppModule. v1 process termination - // closes the gRPC sockets via OS-level descriptor reclaim — fine - // for the dev/preprod posture; prod will add an explicit - // OnApplicationShutdown hook once Nest's shutdown hooks are - // enabled in main.ts. +export class AiClientModule implements OnApplicationShutdown { + constructor( + @Inject(AI_CHAT_GRPC_CLIENT) private readonly chatStub: Client, + @Inject(AI_RAG_GRPC_CLIENT) private readonly ragStub: Client, + @Inject(AI_INGESTION_GRPC_CLIENT) private readonly ingestionStub: Client, + @Inject(AI_MODELS_GRPC_CLIENT) private readonly modelsStub: Client, + ) {} + + /** + * Close every generated gRPC stub when the BFF receives `SIGTERM` + * / `SIGINT`. Each `Client.close()` flushes pending RPCs (with + * their own gRPC `CANCELLED` semantics) and tears down the + * shared HTTP/2 channel so the process can exit promptly without + * waiting for the channel's keepalive PINGs. + * + * The four stubs share the same underlying HTTP/2 channel (same + * endpoint + same credentials, gRPC-js de-duplicates), so the + * four `close()` calls are cheap but kept explicit — adding a + * fifth stub later means adding a fifth `close()` line, which is + * easier to spot than iterating an array that grew without + * review. + */ + onApplicationShutdown(): void { + this.chatStub.close(); + this.ragStub.close(); + this.ingestionStub.close(); + this.modelsStub.close(); + } } diff --git a/apps/portal-bff/src/main.ts b/apps/portal-bff/src/main.ts index abbc33b..0194920 100644 --- a/apps/portal-bff/src/main.ts +++ b/apps/portal-bff/src/main.ts @@ -81,6 +81,16 @@ async function bootstrap() { const app = await NestFactory.create(AppModule, { bufferLogs: true }); app.useLogger(app.get(Logger)); + // Wire `SIGTERM` / `SIGINT` / `SIGHUP` to NestJS lifecycle hooks + // (`OnApplicationShutdown`). Without this call, modules that + // hold long-lived resources (gRPC channels to `apf-ai-service` + // per [ADR-0024](../../../docs/decisions/0024-ai-service-relay-grpc-sse-bridge.md), + // Redis clients, …) would only release them via OS-level + // descriptor reclaim at process death, which delays orderly + // termination on `pnpm nx serve` reload and on prod rolling + // restarts. + app.enableShutdownHooks(); + // Global exception filter — normalises every 4xx/5xx response to // `{ error: { code, message, traceId } }`. The Nest default // serialises HttpException's getResponse() at the top level, diff --git a/docs/decisions/0024-ai-service-relay-grpc-sse-bridge.md b/docs/decisions/0024-ai-service-relay-grpc-sse-bridge.md index ebe6437..c6876e2 100644 --- a/docs/decisions/0024-ai-service-relay-grpc-sse-bridge.md +++ b/docs/decisions/0024-ai-service-relay-grpc-sse-bridge.md @@ -1,5 +1,5 @@ --- -status: proposed +status: accepted date: 2026-05-19 decision-makers: R&D Lead tags: [backend, security, observability] diff --git a/docs/decisions/README.md b/docs/decisions/README.md index 3923a6d..ec2d7f6 100644 --- a/docs/decisions/README.md +++ b/docs/decisions/README.md @@ -67,4 +67,4 @@ ADRs are listed in numerical order. To slice by topic, filter on the `Tags` colu | [0021](0021-phase-2-security-baseline.md) | Phase-2 security baseline — helmet, CORS allowlist, double-submit CSRF, rate limiting, structured error envelope | accepted | `security`, `backend` | 2026-05-13 | | [0022](0022-docs-site-vitepress.md) | Documentation site — VitePress + Mermaid plugin, separate static deployment | accepted | `process`, `infrastructure` | 2026-05-15 | | [0023](0023-charts-d3-observable-plot.md) | Charts + dashboards — D3 + Observable Plot wrapped in `libs/shared/charts` | accepted | `frontend`, `accessibility`, `performance` | 2026-05-16 | -| [0024](0024-ai-service-relay-grpc-sse-bridge.md) | AI service relay — vendored gRPC protos, NestJS gRPC client, SSE bridge to the SPA, POC unsigned principal | proposed | `backend`, `security`, `observability` | 2026-05-19 | +| [0024](0024-ai-service-relay-grpc-sse-bridge.md) | AI service relay — vendored gRPC protos, NestJS gRPC client, SSE bridge to the SPA, POC unsigned principal | accepted | `backend`, `security`, `observability` | 2026-05-19 |