gateway-provider.ts

gateway-provider.tsapps/api/src/modules/ai/gateway-provider.tsraw ← ai33,946 B← Back to the app
1/**
2 * AI module — `AiGateway` provider + budget guard (Phase 12, TASK-0121).
3 *
4 * Wires the `@factory/ai` gateway into the API:
5 *
6 *   - **config resolution** (per workspace, per call): the workspace's
7 *     `settings.ai` (Zod-validated `AiConfigSchema`) over the platform env
8 *     defaults (`AI_PROVIDER`/`AI_BASE_URL`/`AI_API_KEY`/`AI_MODEL`) with
9 *     per-field fallback. Provider selection is configuration-only: swapping
10 *     `AI_PROVIDER` (or a workspace's `settings.ai.provider`) changes zero
11 *     application code (ROADMAP Phase 12 exit criterion).
12 *   - **usage ledger**: every completed (or failed-after-consume) gateway
13 *     call appends an `ai_messages` row — in the caller's conversation when
14 *     the feature has one (chat), else a hidden per-call conversation
15 *     (`title: "ai-usage:<capability>"`). The monthly rollup (the budget
16 *     guard + the `usage.rollup` worker job) sums `ai_messages` per
17 *     workspace since month-start — one ledger, no second table.
18 *   - **budget guard** ({@link createAiBudgetGuard}): the service-level
19 *     guard every AI endpoint awaits BEFORE calling the gateway — a
20 *     Postgres-backed month-to-date total vs
21 *     `workspace.usageLimits.aiMonthlyTokens` (unset/0 = unlimited), a
22 *     Redis fixed-window rate cap, and a 24 h-throttled admin notification
23 *     (Redis `SET … NX EX` slot — globally consistent across replicas) +
24 *     the `ai.budget_exceeded` audit event. The gateway's own
25 *     submit-time check (per-process in-memory ledger) stays as a fast
26 *     backstop.
27 *   - **error translation** ({@link mapAiError}): the gateway throws the
28 *     `@factory/shared` envelope; it is re-raised as the API's `ApiError`
29 *     with the canonical code — the budget 402 surfaces as a TOP-LEVEL
30 *     `AI_BUDGET_EXCEEDED` (the shared package carries it in
31 *     `details.code`; the API's error table is extensible by design —
32 *     recorded decision in `@factory/ai/src/usage.ts` +
33 *     `docs/api/authentication.md` §9.1).
34 *
35 * **Mock provider (CI/dev, recorded decision):** with `AI_PROVIDER=mock`
36 * the registry returns a deterministic `MockProvider` — the CI capability
37 * suite (TASK-0121) and local dev run the full gateway surface offline.
38 * The mock is dev/test configuration, never a production provider value.
39 */
40import {
41  AiConfigSchema,
42  AiGateway,
43  AnthropicAdapter,
44  BudgetService,
45  CustomAdapter,
46  GoogleAdapter,
47  MockProvider,
48  OpenAIAdapter,
49  OssAdapter,
50  resolveAiConfig,
51  resolveKeyRef,
52  type ProviderAdapter,
53  type ProviderId,
54  type ResolvedAiConfig,
55  UsageLedger,
56  type UsageRecord,
57} from "@factory/ai";
58import { ApiError as SharedApiError, ErrorCodes as SharedErrorCodes } from "@factory/shared";
59import { z } from "zod";
60 
61import type { Request } from "express";
62import type { PrismaClient } from "@prisma/client";
63import type Redis from "ioredis";
64import { ApiError } from "../../common/errors.js";
65import { logger } from "../../common/logger.js";
66import { loadEnv } from "../../config/env.js";
67 
68/** DI token for the gateway (the module constructs it once at boot). */
69export const AI_GATEWAY = "AI_GATEWAY";
70/** DI token for the service-level budget guard. */
71export const AI_BUDGET_GUARD = "AI_BUDGET_GUARD";
72 
73/**
74 * The dimension the OpenSearch content index's vector field uses. The
75 * gateway's embed result is checked against it (fail fast with a config
76 * error, not a corrupt index at query time — TASK-0121 acceptance). The
77 * default matches the search package's `DEFAULT_VECTOR_DIMENSION`; the
78 * capability suite sets it to the mock's dimension.
79 */
80export function embedDimension(): number {
81  const raw = process.env.SEARCH_VECTOR_DIMENSION || process.env.AI_EMBEDDING_DIMENSION;
82  const n = raw ? Number.parseInt(raw, 10) : NaN;
83  return Number.isFinite(n) && n > 0 ? n : 1536;
84}
85 
86export interface AiGatewayDeps {
87  /**
88   * Budget-exceeded hook (invoked by the gateway's in-memory backstop when
89   * it trips; the module no-ops it — the service-level guard is the
90   * enforcement + notification point). At most once per workspace per 24 h
91   * (the `BudgetService` throttle).
92   */
93  onBudgetExceeded: (workspaceId: string, used: number, limit: number) => Promise<void> | void;
94  /**
95   * Usage sink: persists a completed/failed gateway call. The module
96   * implements this (ai_messages ledger-row upsert + conversation
97   * bookkeeping).
98   */
99  ledgerSink: (record: UsageRecord) => Promise<void>;
100}
101 
102const envAiProviderSchema = z.enum(["openai", "anthropic", "google", "oss", "custom", "mock"]);
103 
104/**
105 * TASK-0122 (E2E): the mock-provider outage arming (single-shot, PER WORKSPACE).
106 *
107 * The mock is created fresh on every gateway call (the registry factory), so
108 * a RUNNING API process cannot be switched into "fail the next call" via
109 * `process.env` (the E2E suite talks to a live server, unlike the in-process
110 * integration suite). Instead the test-seam endpoint
111 * (`POST /internal/workspaces/:workspaceId/ai/mock-fail`, service-token
112 * guarded) arms the workspace, and the NEXT mock the registry builds FOR THAT
113 * WORKSPACE consumes it (single-shot per workspace).
114 *
115 * Why per-workspace (not process-global): Playwright runs its desktop +
116 * mobile projects IN PARALLEL against the same API process, and each E2E test
117 * uses its own `freshWorkspace`. A process-global arming would be consumed by
118 * whichever request landed first — a happy-path test from the other project
119 * could eat this test's arming (or worse, a happy test could receive the
120 * error). Scoping by workspace makes the arming deterministic regardless of
121 * project concurrency.
122 *
123 * Only meaningful when `AI_PROVIDER=mock`; a no-op for real providers.
124 */
125const mockOutageArmings = new Map<string, { status: number; deltas: number }>();
126 
127/**
128 * Test seam: make the NEXT mock chat call FOR `workspaceId` fail MID-stream
129 * after `deltas` deltas (with a retryable `status`). Exactly ONE request in
130 * that workspace fails — the SSE stream carries the in-band `error` delta +
131 * terminal `done` — and the next request (the UI's Retry button, a fresh
132 * gateway call in the same workspace) succeeds. Returns the armed status.
133 */
134export function armMockFailures(workspaceId: string, status = 500, deltas = 2): number {
135  mockOutageArmings.set(workspaceId, { status, deltas });
136  return status;
137}
138 
139/** Drop any pending arming for a workspace (called by `resetWorkspaceCache`). */
140export function clearMockOutageArming(workspaceId: string): void {
141  mockOutageArmings.delete(workspaceId);
142}
143 
144// TASK-0123 (test seam): make the NEXT mock WRITING call for `workspaceId`
145// append a blockId the prompt never contained — the deterministic seam for
146// the server-side "unknown ID → drop + note, never applied" validation.
147const mockUnknownIdArmings = new Map<string, string>();
148 
149/** Arm the next mock writing call for `workspaceId` to emit `unknownId`. */
150export function armMockUnknownBlockId(workspaceId: string, unknownId: string): void {
151  mockUnknownIdArmings.set(workspaceId, unknownId);
152}
153 
154/** Drop any pending arming for a workspace (called by `resetWorkspaceCache`). */
155export function clearMockUnknownBlockId(workspaceId: string): void {
156  mockUnknownIdArmings.delete(workspaceId);
157}
158 
159// TASK-0124 (test seam): make the NEXT mock DATABASE-GENERATION call for
160// `workspaceId` emit a deliberately INVALID tool call (a non-generatable
161// property type / an unknown propertyId / an unparseable formula) — the
162// deterministic seam for the "zod/engine validation fails → retry once →
163// 422, no partial write" path.
164const mockDbInvalidArmings = new Map<string, string>();
165 
166/** Arm the next mock database call for `workspaceId` to emit an invalid call. */
167export function armMockDbInvalidStructure(workspaceId: string, action: "generate" | "extract" | "formula" = "generate"): void {
168  mockDbInvalidArmings.set(workspaceId, action);
169}
170 
171/** Drop any pending arming for a workspace (called by `resetWorkspaceCache`). */
172export function clearMockDbInvalidStructure(workspaceId: string): void {
173  mockDbInvalidArmings.delete(workspaceId);
174}
175 
176// TASK-0126 (test seam): make the NEXT mock PAGE-GENERATION call for
177// `workspaceId` emit a failing plan step (`failStep`, a 1-based step number
178// whose injected step references an unknown ref → the executor's rollback
179// path) and/or fetch the plan slowly (`slowMs`, so the service's total-timeout
180// watchdog fires → the `ai_timeout` path). Both are armed via the internal
181// test-seam endpoint (single-shot PER WORKSPACE, consumed by the next generate
182// chat call built for that workspace).
183const mockGenerateArmings = new Map<string, { failStep?: string; slowMs?: string }>();
184 
185/** Arm the next mock generate call for `workspaceId` (the timeout/rollback seams). */
186export function armMockGenerateSeam(
187  workspaceId: string,
188  opts: { failStep?: number; slowMs?: number },
189): { failStep?: string; slowMs?: string } {
190  const arming = mockGenerateArmings.get(workspaceId) ?? {};
191  if (opts.failStep !== undefined) arming.failStep = String(opts.failStep);
192  if (opts.slowMs !== undefined) arming.slowMs = String(opts.slowMs);
193  mockGenerateArmings.set(workspaceId, arming);
194  return arming;
195}
196 
197/** Drop any pending arming for a workspace (called by `resetWorkspaceCache`). */
198export function clearMockGenerateSeam(workspaceId: string): void {
199  mockGenerateArmings.delete(workspaceId);
200}
201 
202// T-172…T-175 (test seam): make the assistant mock calls for `workspaceId`
203// emit a deliberately INVALID tool call (`hostile` → the "validate → retry
204// once → 422, no write" path) and/or a forged candidate id (`forgeId` → the
205// "no fabricated refs → drop, never use" path). PERSISTENT per workspace (like
206// the database-invalid arming: the retry-once makes two chat calls for the
207// same capability, so the arming must survive the first). Cleared by
208// `resetWorkspaceCache`.
209const mockAssistantArmings = new Map<string, { hostile: boolean; forgeId: boolean }>();
210 
211/** Arm the assistant mock for `workspaceId` (hostile and/or forgeId). */
212export function armMockAssistantSeam(workspaceId: string, opts: { hostile?: boolean; forgeId?: boolean }): void {
213  const arming = mockAssistantArmings.get(workspaceId) ?? { hostile: false, forgeId: false };
214  if (opts.hostile !== undefined) arming.hostile = opts.hostile;
215  if (opts.forgeId !== undefined) arming.forgeId = opts.forgeId;
216  mockAssistantArmings.set(workspaceId, arming);
217}
218 
219/** Drop any pending arming for a workspace (called by `resetWorkspaceCache`). */
220export function clearMockAssistantSeam(workspaceId: string): void {
221  mockAssistantArmings.delete(workspaceId);
222}
223 
224/**
225 * The platform-level `maxOutputTokens` default (env `AI_MAX_OUTPUT_TOKENS`).
226 * Unset → the gateway's 4096 default (no behavior change for other
227 * deployments). See the recorded decision in `resolveConfig`.
228 */
229function platformMaxOutputTokens(): number {
230  const n = Number(process.env.AI_MAX_OUTPUT_TOKENS);
231  return Number.isFinite(n) && n > 0 ? Math.floor(n) : 4096;
232}
233 
234/**
235 * Adapter stream timeouts (env-overridable; defaults = the adapter's built-in
236 * values, so unset env changes nothing):
237 *
238 *   - `AI_STREAM_IDLE_MS` — the max gap between streamed BYTES before the
239 *     stream aborts (adapter default 30 s). Recorded decision (2025): the OSS
240 *     reasoning model streams a long SILENT thinking phase (no tool-call
241 *     deltas until the plan JSON starts) — a 39 s gap was observed between the
242 *     last thinking byte and the first plan byte, which the 30 s default turns
243 *     into a mid-stream 504 the gateway cannot retry (partial output
244 *     already yielded). 90 s covers the observed silence with margin without
245 *     hiding a genuinely dead stream.
246 *   - `AI_STREAM_TIMEOUT_MS` — the fetch-level cap (adapter default 60 s;
247 *     covers the response phase, not the whole stream).
248 */
249function adapterTimeouts(): { timeoutMs?: number; idleMs?: number } {
250  const t: { timeoutMs?: number; idleMs?: number } = {};
251  const to = Number(process.env.AI_STREAM_TIMEOUT_MS);
252  if (Number.isFinite(to) && to > 0) t.timeoutMs = Math.floor(to);
253  const idle = Number(process.env.AI_STREAM_IDLE_MS);
254  if (Number.isFinite(idle) && idle > 0) t.idleMs = Math.floor(idle);
255  return t;
256}
257 
258/** The env platform defaults (the per-field fallback floor). */
259function envDefaults(): {
260  provider: "openai" | "anthropic" | "google" | "oss" | "custom";
261  chatModel: string;
262  embedModel: string;
263  baseUrl?: string;
264  apiKeyRef?: string;
265} {
266  const env = loadEnv();
267  const parsed = envAiProviderSchema.safeParse(env.AI_PROVIDER);
268  // "mock" resolves to the "custom" family; the registry inspects the raw
269  // env to swap in the MockProvider (configuration, not code).
270  const provider = !parsed.success || parsed.data === "mock" ? "custom" : parsed.data;
271  return {
272    provider,
273    chatModel: env.AI_MODEL || "gpt-4o",
274    embedModel: process.env.AI_EMBEDDING_MODEL || "text-embedding-3-small",
275    ...(env.AI_BASE_URL ? { baseUrl: env.AI_BASE_URL } : {}),
276    ...(env.AI_API_KEY ? { apiKeyRef: "env:AI_API_KEY" } : {}),
277  };
278}
279 
280/**
281 * Resolve the API key for a resolved config. `secretsRef` is a NAME, never a
282 * literal: `env:NAME` resolves from the process environment; a bare name
283 * resolves against the provider's env key (this phase's secret store is the
284 * environment — the encrypted `provider_secrets` store is enterprise
285 * scope). Raw key values never appear in config, logs, or audits.
286 */
287function apiKeyFor(config: ResolvedAiConfig): string | undefined {
288  const ref = config.secretsRef;
289  if (!ref) return undefined;
290  if (ref.startsWith("env:")) return resolveKeyRef(ref, process.env as Record<string, string | undefined>);
291  return process.env.AI_API_KEY;
292}
293 
294export interface WorkspaceAiRow {
295  settings: unknown;
296  usageLimits: unknown;
297  billingPlan: string;
298}
299 
300/**
301 * The guard's budget for a workspace: `usageLimits.aiMonthlyTokens` (the
302 * admin override — the task's `aiMonthlyTokens` field) wins; the workspace
303 * `settings.ai.budgets.tokensPerMonth` is the config-level default.
304 * Unset / 0 = unlimited (all current plans).
305 */
306export function tokensPerMonthFor(row: WorkspaceAiRow, settingsAi: { budgets?: { tokensPerMonth?: number } } | null): number {
307  const usageLimits = (row.usageLimits ?? {}) as { aiMonthlyTokens?: unknown };
308  const fromLimits =
309    typeof usageLimits.aiMonthlyTokens === "number" && Number.isFinite(usageLimits.aiMonthlyTokens)
310      ? Math.max(0, usageLimits.aiMonthlyTokens)
311      : 0;
312  const fromSettings =
313    typeof settingsAi?.budgets?.tokensPerMonth === "number" ? Math.max(0, settingsAi.budgets.tokensPerMonth) : 0;
314  return Math.max(fromLimits, fromSettings);
315}
316 
317/**
318 * Build the gateway for the module. The gateway's own submit-time guard
319 * uses its (per-process) in-memory ledger as a fast backstop; the
320 * Postgres-backed, replica-consistent enforcement is the service-level
321 * {@link AiBudgetGuard} every AI endpoint awaits before the gateway call.
322 */
323export function buildAiGateway(deps: AiGatewayDeps): AiGateway {
324  const ledger = new UsageLedger({
325    sink: async (record: UsageRecord) => {
326      try {
327        await deps.ledgerSink(record);
328      } catch (err) {
329        // The ledger must never take a request down: usage loss is logged,
330        // the nightly rollup reconciles.
331        logger.warn(`ai usage sink failed (${record.workspaceId}): ${(err as Error).message}`);
332      }
333    },
334  });
335 
336  const budgets = new BudgetService(ledger);
337 
338  const resolveConfig = (workspaceId: string): ResolvedAiConfig => {
339    const row = cachedWorkspaceRow(workspaceId);
340    if (!row) {
341      // Services prime the cache before any gateway call (a cold miss means
342      // the workspace vanished between prime and call — 404).
343      throw new SharedApiError(SharedErrorCodes.NOT_FOUND, { message: "Workspace not found.", status: 404 });
344    }
345    const settings = (row.settings ?? {}) as { ai?: unknown };
346    const parsed = AiConfigSchema.safeParse(settings.ai ?? {});
347    const workspaceCfg = parsed.success ? parsed.data : {};
348    const config = resolveAiConfig(workspaceCfg, null, envDefaults());
349    // The registry reads `workspaceId` to scope the per-workspace mock-outage
350    // arming (test seam) — see `mockOutageArmings` above.
351    config.workspaceId = workspaceId;
352    config.budgets = {
353      ...config.budgets,
354      tokensPerMonth: tokensPerMonthFor(row, parsed.success ? parsed.data : null),
355      // Platform-level output-token default (env `AI_MAX_OUTPUT_TOKENS`).
356      //
357      // Recorded decision (2025, root cause of a silent plan truncation): the
358      // gateway clamps every chat request to `budgets.maxOutputTokens`
359      // (default 4096). The OSS reasoning model this deployment runs
360      // (qwen3.8-27B via vLLM) emits ~9 k REASONING tokens before a complex
361      // generation plan of ~6 k more — a 4096 clamp cuts the plan JSON
362      // mid-stream, the arguments no longer parse, and the run dies. The env
363      // var raises the platform default for deployments that need it (unset →
364      // the 4096 default, so other deployments are untouched); a workspace's
365      // own explicit `settings.ai.budgets.maxOutputTokens` still wins.
366      maxOutputTokens:
367        parsed.success &&
368        (parsed.data as { budgets?: { maxOutputTokens?: number } }).budgets?.maxOutputTokens != null
369          ? (config.budgets.maxOutputTokens)
370          : platformMaxOutputTokens(),
371    };
372    return config;
373  };
374 
375  const registry = {
376    get(providerId: ProviderId, config: ResolvedAiConfig): ProviderAdapter {
377      const env = loadEnv();
378      const key = apiKeyFor(config);
379      const baseUrl = config.endpoint?.baseUrl ?? env.AI_BASE_URL ?? "https://api.openai.com/v1";
380      // CI/dev mock (recorded decision): the capability suite runs the whole
381      // gateway surface offline. Deviates from the production provider list
382      // on purpose — it is a test adapter, selected by test config only.
383      if (env.AI_PROVIDER === "mock") {
384        const mock = new MockProvider(providerId === "anthropic" ? "anthropic" : "custom", embedDimension());
385        // TASK-0122: optional per-delta pacing so mid-stream-cancellation
386        // tests can land an abort BETWEEN deltas (0 = instant, the default).
387        const delay = Number(process.env.AI_MOCK_DELTA_DELAY_MS);
388        if (Number.isFinite(delay) && delay > 0) mock.deltaDelayMs = delay;
389        // TASK-0122 (E2E): arm a deterministic provider outage FOR THIS
390        // WORKSPACE. The mock is created PER gateway call (the registry
391        // factory), so a RUNNING API process cannot be flipped via
392        // process.env — the test-seam endpoint
393        // (`POST /internal/workspaces/:workspaceId/ai/mock-fail`) arms the
394        // workspace and the next mock built for it consumes the arming
395        // (single-shot PER WORKSPACE — see `mockOutageArmings`). Scoping by
396        // workspace (via `config.workspaceId`) keeps it deterministic even
397        // when Playwright's desktop + mobile projects hit the API in parallel.
398        //
399        // The failure is MID-stream (a few deltas, then a retryable 5xx):
400        // the gateway never re-streams partial output (it would duplicate
401        // content), so the SSE endpoint emits an in-band `error` delta +
402        // terminal `done {reason:"error"}` → the panel's retryable banner.
403        // A PRE-stream failure would be useless here: the gateway's retry
404        // loop reuses the same adapter instance, so it auto-recovers inside
405        // the request and the user never sees an error.
406        if (config.workspaceId) {
407          const armed = mockOutageArmings.get(config.workspaceId);
408          if (armed && armed.deltas > 0) {
409            mock.failAfterDeltas = armed.deltas;
410            mock.failStatus = armed.status;
411            mockOutageArmings.delete(config.workspaceId); // single-shot
412          }
413          // TASK-0123 seam: the next WRITING call emits a forged blockId
414          // (single-shot per workspace, like the outage arming).
415          const unknownId = mockUnknownIdArmings.get(config.workspaceId);
416          if (unknownId) {
417            mockUnknownIdArmings.delete(config.workspaceId); // single-shot
418            const originalChat = mock.chat.bind(mock);
419            mock.chat = (req: Parameters<typeof originalChat>[0]) =>
420              originalChat({
421                ...req,
422                metadata: { ...req.metadata, ...(req.metadata?.mode === "writing" ? { forceUnknownId: unknownId } : {}) },
423              });
424          }
425          // TASK-0124 seam: the DATABASE-GENERATION calls for `workspaceId`
426          // emit a deliberately INVALID tool call — the "validation fails →
427          // retry once → 422, no partial write" path.
428          //
429          // PERSISTENT (unlike the single-shot outage / unknown-id armings):
430          // the retry-once makes TWO chat calls for the SAME action, so the
431          // arming must survive the first call for the 422 to actually fire.
432          // It is scoped to the armed ACTION (generate/extract/formula), so a
433          // hostile `generate` does not poison a later `extract` in the same
434          // workspace. Tests clear it via `resetWorkspaceCache` / a fresh
435          // workspace.
436          const dbInvalidAction = mockDbInvalidArmings.get(config.workspaceId);
437          if (dbInvalidAction) {
438            const originalChat2 = mock.chat.bind(mock);
439            mock.chat = (req: Parameters<typeof originalChat2>[0]) =>
440              originalChat2({
441                ...req,
442                metadata: { ...req.metadata, ...(req.metadata?.mode === "database" ? { hostile: dbInvalidAction } : {}) },
443              });
444          }
445          // TASK-0126 seam: the next PAGE-GENERATION call emits the armed
446          // failing plan step and/or the armed plan latency (single-shot per
447          // workspace, like the outage arming — generate is one chat call per
448          // run, so no persistence is needed).
449          const genSeam = mockGenerateArmings.get(config.workspaceId);
450          if (genSeam) {
451            mockGenerateArmings.delete(config.workspaceId); // single-shot
452            const originalChat3 = mock.chat.bind(mock);
453            mock.chat = (req: Parameters<typeof originalChat3>[0]) =>
454              originalChat3({
455                ...req,
456                metadata: {
457                  ...req.metadata,
458                  ...(req.metadata?.mode === "generate"
459                    ? { ...(genSeam.failStep ? { generateFailStep: genSeam.failStep } : {}), ...(genSeam.slowMs ? { generateSlowMs: genSeam.slowMs } : {}) }
460                    : {}),
461                },
462              });
463          }
464          // T-172…T-175 seam: the ASSISTANT calls for `config.workspaceId`
465          // emit a hostile (invalid) tool call and/or a forged candidate id —
466          // persistent, like the database-invalid arming (retry-once).
467          const assistantSeam = mockAssistantArmings.get(config.workspaceId);
468          if (assistantSeam) {
469            const originalChat4 = mock.chat.bind(mock);
470            mock.chat = (req: Parameters<typeof originalChat4>[0]) =>
471              originalChat4({
472                ...req,
473                metadata: {
474                  ...req.metadata,
475                  ...(req.metadata?.mode === "assistant"
476                    ? {
477                        ...(assistantSeam.hostile ? { hostile: "assistant" } : {}),
478                        ...(assistantSeam.forgeId ? { forgeId: "1" } : {}),
479                      }
480                    : {}),
481                },
482              });
483          }
484        }
485        return mock;
486      }
487      // Env-tunable stream timeouts for every real adapter (see
488      // `adapterTimeouts` — defaults unchanged when the env is unset).
489      const timeouts = adapterTimeouts();
490      switch (providerId) {
491        case "openai":
492          return new OpenAIAdapter({ baseUrl, ...timeouts, ...(key ? { apiKey: key } : {}) });
493        case "anthropic":
494          return new AnthropicAdapter({ ...timeouts, ...(key ? { apiKey: key } : { apiKey: "" }) });
495        case "google":
496          return new GoogleAdapter({ ...timeouts, ...(key ? { apiKey: key } : { apiKey: "" }) });
497        case "oss":
498          return new OssAdapter({ baseUrl, ...timeouts, ...(key ? { apiKey: key } : {}) });
499        case "custom":
500          return new CustomAdapter({ baseUrl, ...timeouts, ...(key ? { apiKey: key } : {}), extraHeaders: config.endpoint?.extraHeaders });
501      }
502    },
503  };
504 
505  return new AiGateway({ resolveConfig, registry, budgets, ledger });
506}
507 
508// ---------------------------------------------------------------------------
509// Workspace-row microtask cache (the sync resolveConfig contract)
510// ---------------------------------------------------------------------------
511 
512const workspaceRowCache = new Map<string, WorkspaceAiRow>();
513const cacheTimestamps = new Map<string, number>();
514const WORKSPACE_CACHE_TTL_MS = 30_000;
515 
516/**
517 * The synchronous `resolveConfig` callback needs the workspace row without an
518 * await: every AI service primes this cache (30 s TTL) before its gateway
519 * call. A cold miss in `resolveConfig` → 404 (the service already checked
520 * membership, so a cold miss means the workspace vanished mid-request).
521 */
522export async function primeWorkspaceCache(prisma: PrismaClient, workspaceId: string): Promise<WorkspaceAiRow> {
523  const hit = workspaceRowCache.get(workspaceId);
524  if (hit && Date.now() - (cacheTimestamps.get(workspaceId) ?? 0) < WORKSPACE_CACHE_TTL_MS) return hit;
525  const row = await prisma.workspace.findUnique({
526    where: { id: workspaceId },
527    select: { settings: true, usageLimits: true, billingPlan: true },
528  });
529  if (!row) throw ApiError.notFound("Workspace not found.");
530  const wsRow = row as unknown as WorkspaceAiRow;
531  workspaceRowCache.set(workspaceId, wsRow);
532  cacheTimestamps.set(workspaceId, Date.now());
533  return wsRow;
534}
535 
536/** Test hook: drop the cache (new budget/settings take effect immediately). */
537export function resetWorkspaceCache(): void {
538  workspaceRowCache.clear();
539  cacheTimestamps.clear();
540  mockOutageArmings.clear();
541  mockUnknownIdArmings.clear();
542  mockDbInvalidArmings.clear();
543  mockGenerateArmings.clear();
544  mockAssistantArmings.clear();
545}
546 
547function cachedWorkspaceRow(workspaceId: string): WorkspaceAiRow | null {
548  if (Date.now() - (cacheTimestamps.get(workspaceId) ?? 0) >= WORKSPACE_CACHE_TTL_MS) {
549    workspaceRowCache.delete(workspaceId);
550    cacheTimestamps.delete(workspaceId);
551  }
552  return workspaceRowCache.get(workspaceId) ?? null;
553}
554 
555// ---------------------------------------------------------------------------
556// Service-level budget guard (Postgres-backed, replica-consistent)
557// ---------------------------------------------------------------------------
558 
559export interface AiBudgetGuard {
560  /**
561   * Await BEFORE any gateway call. Throws (shared envelope, mapped by
562   * {@link mapAiError} at the service boundary):
563   *   - `RATE_LIMITED` (429) — per-workspace 60 s fixed window over Redis;
564   *   - `AI_BUDGET_EXCEEDED` (402) — month-to-date ≥ the monthly budget
565   *     (estimating this call's output at 2×maxOutputTokens, arch §7.2).
566   * On budget exhaustion the admin notification + audit fire (throttled to
567   * one per workspace per 24 h via a Redis `SET NX EX` slot).
568   */
569  check(workspaceId: string, maxOutputTokens?: number, req?: Request): Promise<void>;
570  /** Month-to-date tokens (for the ping diagnostic + dashboards). */
571  usedTokens(workspaceId: string): Promise<number>;
572  /** The resolved monthly budget for the workspace (0 = unlimited). */
573  monthlyBudget(workspaceId: string): Promise<number>;
574}
575 
576export function createAiBudgetGuard(opts: {
577  prisma: PrismaClient;
578  redis: Redis;
579  onBudgetExceeded: (workspaceId: string, used: number, limit: number, req?: Request) => Promise<void> | void;
580}): AiBudgetGuard {
581  const { prisma, redis, onBudgetExceeded } = opts;
582 
583  function monthStart(): Date {
584    const d = new Date();
585    return new Date(d.getFullYear(), d.getMonth(), 1);
586  }
587 
588  async function usedTokens(workspaceId: string): Promise<number> {
589    const rows = await prisma.aiMessage.aggregate({
590      where: { conversation: { workspaceId }, createdAt: { gte: monthStart() } },
591      _sum: { tokensIn: true, tokensOut: true },
592    });
593    return (rows._sum.tokensIn ?? 0) + (rows._sum.tokensOut ?? 0);
594  }
595 
596  async function monthlyBudget(workspaceId: string): Promise<number> {
597    const row = await primeWorkspaceCache(prisma, workspaceId);
598    const settings = (row.settings ?? {}) as { ai?: unknown };
599    const parsed = AiConfigSchema.safeParse(settings.ai ?? {});
600    return tokensPerMonthFor(row, parsed.success ? parsed.data : null);
601  }
602 
603  async function check(workspaceId: string, maxOutputTokens = 4096, req?: Request): Promise<void> {
604    const row = await primeWorkspaceCache(prisma, workspaceId);
605    const settings = (row.settings ?? {}) as { ai?: unknown };
606    const parsed = AiConfigSchema.safeParse(settings.ai ?? {});
607    const cfg = parsed.success ? parsed.data : null;
608    const limit = tokensPerMonthFor(row, cfg);
609    const requestsPerMinute = typeof cfg?.budgets?.requestsPerMinute === "number" ? cfg.budgets.requestsPerMinute : 60;
610 
611    // 1) rate window (per-workspace 60 s fixed window over Redis; the
612    //    per-USER 30 msg/min edge cap is the AI limiter — TASK-0122).
613    if (requestsPerMinute > 0) {
614      const minute = Math.floor(Date.now() / 60_000);
615      const key = `ai:rl:ws:${workspaceId}:${minute}`;
616      let count = 0;
617      try {
618        count = (await redis.incr(key)) as number;
619        if (count === 1) await redis.expire(key, 120);
620      } catch (err) {
621        // Redis blip: degrade open (the per-process gateway backstop still
622        // rate-limits; the API never crashes on a Redis failure).
623        logger.warn(`ai rate window unreadable (${workspaceId}): ${(err as Error).message}`);
624      }
625      if (count > requestsPerMinute) {
626        throw new SharedApiError(SharedErrorCodes.RATE_LIMITED, {
627          message: "AI rate limit reached. Try again shortly.",
628          details: { code: "RATE_LIMITED", limit: requestsPerMinute, windowSeconds: 60 },
629        });
630      }
631    }
632 
633    // 2) monthly token budget (0 = unlimited). The estimate (2×
634    //    maxOutputTokens, arch §7.2) keeps a huge generation from crossing
635    //    the line before the stream starts.
636    if (limit > 0) {
637      const used = await usedTokens(workspaceId);
638      if (used + 2 * maxOutputTokens >= limit) {
639        if (await acquireNotificationSlot(redis, workspaceId)) {
640          try {
641            await onBudgetExceeded(workspaceId, used, limit, req);
642          } catch (err) {
643            logger.warn(`ai budget notification failed (${workspaceId}): ${(err as Error).message}`);
644          }
645        }
646        throw new SharedApiError(SharedErrorCodes.AI_ERROR, {
647          status: 402,
648          message: "Monthly AI token budget reached. Ask an admin.",
649          details: { code: "AI_BUDGET_EXCEEDED", reason: "monthly_token_budget_exceeded", used, limit },
650        });
651      }
652    }
653  }
654 
655  return { check, usedTokens, monthlyBudget };
656}
657 
658/**
659 * The 24 h notification throttle: a Redis `SET NX EX 86400` slot (atomic,
660 * replica-consistent). Redis failure → allow the notification (fail open:
661 * a duplicate email is preferable to a missed budget alert).
662 */
663async function acquireNotificationSlot(redis: Redis, workspaceId: string): Promise<boolean> {
664  const key = `ai:budget:notified:${workspaceId}`;
665  try {
666    const ok = await redis.set(key, "1", "EX", 24 * 3600, "NX");
667    return ok === "OK";
668  } catch {
669    return true;
670  }
671}
672 
673// ---------------------------------------------------------------------------
674// Shared-envelope → API ApiError translation
675// ---------------------------------------------------------------------------
676 
677const SHARED_CODE_TO_API: Record<string, string> = {
678  VALIDATION_ERROR: "VALIDATION_ERROR",
679  UNAUTHORIZED: "UNAUTHORIZED",
680  FORBIDDEN: "FORBIDDEN",
681  NOT_FOUND: "NOT_FOUND",
682  CONFLICT: "CONFLICT",
683  RATE_LIMITED: "RATE_LIMITED",
684  NOT_IMPLEMENTED: "NOT_IMPLEMENTED",
685  SEARCH_UNAVAILABLE: "SERVICE_UNAVAILABLE",
686  AI_ERROR: "SERVICE_UNAVAILABLE",
687  TIMEOUT: "SERVICE_UNAVAILABLE",
688  INTERNAL: "INTERNAL_ERROR",
689};
690 
691/**
692 * Translate a gateway/guard throw into the API's ApiError. The budget 402
693 * arrives as shared `AI_ERROR` + `details.code = AI_BUDGET_EXCEEDED` (the
694 * frozen shared enum has no 402 code) — it is re-raised as a TOP-LEVEL
695 * `AI_BUDGET_EXCEEDED` (the API's canonical table, extensible by design).
696 */
697export function mapAiError(err: unknown): ApiError {
698  if (err instanceof ApiError) return err;
699  const shared = err as Partial<SharedApiError> | null;
700  if (shared && typeof shared.code === "string" && typeof shared.status === "number") {
701    const details =
702      shared.details && typeof shared.details === "object" && !Array.isArray(shared.details)
703        ? [shared.details as Record<string, unknown>]
704        : undefined;
705    const budget = details?.find((d) => d?.code === "AI_BUDGET_EXCEEDED") as
706      | { code: string; reason?: string; used?: number; limit?: number }
707      | undefined;
708    if (budget) {
709      return new ApiError("AI_BUDGET_EXCEEDED", 402, shared.message ?? "AI budget exceeded.", [
710        { reason: budget.reason ?? "monthly_token_budget_exceeded", used: budget.used ?? 0, limit: budget.limit ?? 0 },
711      ]);
712    }
713    const code = SHARED_CODE_TO_API[shared.code];
714    if (!code) {
715      logger.warn(`unmapped AI gateway error code: ${shared.code}`);
716      return ApiError.internal("AI request failed.");
717    }
718    return new ApiError(code as ApiError["code"], shared.status, shared.message ?? "AI request failed.", details);
719  }
720  logger.warn(`unmapped AI gateway error: ${err instanceof Error ? err.message : String(err)}`);
721  return ApiError.internal("AI request failed.");
722}
723