database-ai.service.ts

database-ai.service.tsapps/api/src/modules/ai/database-ai.service.tsraw ← ai36,514 B← Back to the app
1/**
2 * TASK-0124 — Database AI (generate database / extract data / generate formula).
3 *
4 * Endpoints (documented in `docs/api/ai.md`):
5 *
6 *   POST /workspaces/:workspaceId/ai/databases/generate     (SSE)
7 *       body: { prompt, parentId? }. Frames: `tool_call` (the validated
8 *       `createDatabase` call), `usage`, `error`, `done`. Pre-stream failures
9 *       (permission 403, rate 429, budget 402) are a canonical JSON envelope;
10 *       in-stream failures (invalid structure after the retry-once, provider
11 *       outage) are in-band `error` + terminal `done {reason:"error"}`.
12 *       `done` on success carries `{ databaseId, propertyIds, recordIds,
13 *       tokensIn, tokensOut, assistantMessageId }`.
14 *
15 *   POST /workspaces/:workspaceId/ai/databases/extract      (SSE)
16 *       body: { databaseId, text (≤50k) }. The model emits `insertRecords`
17 *       (propertyId-keyed, ≤200). Unknown markdown columns → `ignoredColumns`
18 *       (in the `done` frame).
19 *
20 *   POST /workspaces/:workspaceId/ai/databases/extract/preview  (JSON, no write)
21 *       body: { databaseId, text } → the deterministic column→property mapping
22 *       + row count (the UI's live preview; no LLM call, no budget).
23 *
24 *   POST /workspaces/:workspaceId/ai/databases/formulas     (JSON, non-streamed)
25 *       body: { databaseId, description, name?, sampleRowIds? } → the model
26 *       emits `generateFormula`; the expression is validated by the safe
27 *       formula engine (parse + type-check + eval on ≤5 rows) BEFORE saving.
28 *       Invalid after the retry-once → 422 `formula_invalid` (no Formula row).
29 *
30 * Security (spec §Security, AGENTS #7/18/19):
31 *   - every write goes through the normal domain services (permissions,
32 *     audit, realtime, search) — never raw SQL, never elevated permissions;
33 *   - the tool-call output is UNTRUSTED: validated by zod before use;
34 *   - formulas are only parsed/evaluated by the safe AST engine (zero eval,
35 *     AGENTS #18);
36 *   - hidden per-call conversations (`db:generate` / `db:extract` /
37 *     `db:formula` titles, `metadata.feature` db.*) never list in the panel.
38 */
39import { Inject, Injectable } from "@nestjs/common";
40import type { Request, Response } from "express";
41 
42import type { PrismaClient } from "@prisma/client";
43import { PrismaService } from "../../common/prisma.service.js";
44import { AuditService } from "../../common/audit.service.js";
45import { ApiError, type ErrorCode, type ErrorDetail } from "../../common/errors.js";
46import { logger } from "../../common/logger.js";
47import { assertCan } from "../../lib/authorize.js";
48import { AI_GATEWAY, mapAiError, primeWorkspaceCache, type AiBudgetGuard } from "./gateway-provider.js";
49import { AiBudgetGuardService } from "./budget-guard.service.js";
50import { AiRateLimitService } from "./ai-rate-limit.js";
51import {
52  type AiExtractInput,
53  type AiExtractPreviewInput,
54  type AiGenerateDatabaseInput,
55  type AiGenerateFormulaInput,
56} from "./database-ai.dto.js";
57import {
58  CREATE_DATABASE_TOOL,
59  DB_EXTRACT_CONV_PREFIX,
60  DB_FORMULA_CONV_PREFIX,
61  DB_FORMULA_SYSTEM_PROMPT,
62  DB_GEN_CONV_PREFIX,
63  DB_GEN_SYSTEM_PROMPT,
64  DB_EXTRACT_SYSTEM_PROMPT,
65  GENERATE_FORMULA_TOOL,
66  INSERT_RECORDS_TOOL,
67  renderExtractPrompt,
68  renderFormulaPrompt,
69  renderGeneratePrompt,
70  validateCreateDatabase,
71  validateInsertRecords,
72  previewExtractMapping,
73  type DbPropMeta,
74} from "./db-prompts.js";
75import { lowerPropertyType, DB_FORMULA_MAX_SAMPLE_ROWS } from "@factory/ai";
76import { DatabaseService } from "../databases/database.service.js";
77import { DatabaseQueryService } from "../databases/query.service.js";
78import { PageService } from "../pages/page.service.js";
79import type { AiGateway } from "@factory/ai";
80 
81const TIMEOUT_MS = 60_000;
82 
83export type AiDatabaseSseEvent =
84  | { type: "delta"; text: string }
85  | { type: "tool_call"; id: string; name: string; arguments: Record<string, unknown> }
86  | { type: "usage"; tokensIn: number; tokensOut: number }
87  | { type: "error"; code: string; message: string; reason?: string; retryable?: boolean }
88  | {
89      type: "done";
90      assistantMessageId?: string;
91      reason?: "ok" | "aborted" | "error";
92      databaseId?: string;
93      propertyIds?: string[];
94      recordIds?: string[];
95      ignoredColumns?: string[];
96      formulaId?: string;
97      expression?: string;
98      returnType?: string;
99      explanation?: string;
100      samples?: Array<{ recordId: string; value: unknown; error?: string }>;
101      tokensIn?: number;
102      tokensOut?: number;
103    };
104 
105function sseFrame(event: AiDatabaseSseEvent): string {
106  return `data: ${JSON.stringify(event)}\n\n`;
107}
108 
109/** A record's stored property values → the formula engine's name→value map. */
110function recordToFormulaProps(
111  properties: Array<{ property: { id: string; name: string; type: string; config: unknown }; value: unknown }>,
112): Record<string, unknown> {
113  const out: Record<string, unknown> = {};
114  for (const rp of properties) {
115    const { name, type, config } = rp.property;
116    const v = rp.value;
117    const cfg = (config ?? {}) as { options?: Array<{ id?: string; name?: string }> };
118    const optName = (id: unknown): unknown => cfg.options?.find((o) => o.id === id)?.name ?? id;
119    switch (type) {
120      case "TEXT":
121      case "RICH_TEXT":
122        out[name] = typeof v === "object" && v !== null && "text" in v ? String((v as { text?: unknown }).text ?? "") : v == null ? null : String(v);
123        break;
124      case "NUMBER":
125        out[name] = typeof v === "object" && v !== null && "value" in v ? (v as { value: unknown }).value ?? null : v;
126        break;
127      case "SELECT":
128      case "STATUS":
129        out[name] = typeof v === "object" && v !== null && "optionId" in v ? optName((v as { optionId: unknown }).optionId) : null;
130        break;
131      case "MULTI_SELECT":
132        out[name] = typeof v === "object" && v !== null && Array.isArray((v as { optionIds?: unknown[] }).optionIds) ? (v as { optionIds: unknown[] }).optionIds.map(optName) : [];
133        break;
134      case "CHECKBOX":
135        out[name] = typeof v === "object" && v !== null && "value" in v ? Boolean((v as { value: unknown }).value) : Boolean(v);
136        break;
137      case "DATE":
138      case "DATE_RANGE":
139        out[name] = typeof v === "object" && v !== null && "start" in v ? (v as { start: unknown }).start ?? null : v;
140        break;
141      case "PERSON":
142        out[name] = typeof v === "object" && v !== null && Array.isArray((v as { userIds?: unknown[] }).userIds) ? (v as { userIds: unknown[] }).userIds : [];
143        break;
144      case "EMAIL":
145      case "PHONE":
146      case "URL":
147        out[name] = typeof v === "object" && v !== null && "value" in v ? (v as { value: unknown }).value ?? null : v;
148        break;
149      case "FILES":
150        out[name] = typeof v === "object" && v !== null && Array.isArray((v as { fileIds?: unknown[] }).fileIds) ? (v as { fileIds: unknown[] }).fileIds : [];
151        break;
152      default:
153        out[name] = null;
154    }
155  }
156  return out;
157}
158 
159/** A stable, short expression hash for audit (never the full expression). */
160function hashExpression(expr: string): string {
161  let h = 5381;
162  for (let i = 0; i < expr.length; i++) h = ((h << 5) + h + expr.charCodeAt(i)) | 0;
163  return `djb2:${(h >>> 0).toString(16)}`;
164}
165 
166/**
167 * The safe formula engine, loaded at RUNTIME (see `loadDatabaseEngine`).
168 * Structural (not the package's types) so the engine's ESM source never
169 * enters the API's commonjs/node10 type graph.
170 */
171interface ValidatedFormulaLike {
172  ast: unknown;
173  returnType: string;
174  dependencies: string[];
175  usesNow: boolean;
176}
177interface FormulaErrorLike {
178  code: string;
179  message: string;
180}
181interface DatabaseEngineModule {
182  validateFormula(src: string, properties: Record<string, unknown>, propertyTypes?: Record<string, string>): ValidatedFormulaLike;
183  evaluateFormula(ast: unknown, ctx: { now: number; properties: Record<string, unknown>; propertyTypes?: Record<string, string> }): { value: { v: unknown; t: string }; steps: number };
184  FormulaError: new (code: string, message?: string) => FormulaErrorLike;
185}
186 
187/**
188 * Load the safe formula engine (`packages/database-engine`) at runtime.
189 *
190 * A static import is impossible here: the engine is a source-first ESM
191 * package (`"type":"module"`, `main` → `./src/*.ts`) whose source does not
192 * compile under the API's commonjs/node10 tsconfig — a static import would
193 * drag the whole engine into the API's type graph. The runtime file-URL
194 * import skips compile-time resolution entirely (same approach as the
195 * legacy `loadFormulaEngine` in `database.service.ts`). AGENTS #18: the
196 * engine is a safe AST parser/evaluator — zero `eval`/`new Function`.
197 *
198 * Returns `undefined` when the engine source is not resolvable (e.g. a
199 * plain-`node` production runtime without a built engine) — the formula
200 * endpoint then answers 503 `formula_engine_unavailable` (recorded decision).
201 */
202async function loadDatabaseEngine(): Promise<DatabaseEngineModule | undefined> {
203  const { existsSync } = await import("node:fs");
204  const { join, resolve } = await import("node:path");
205  const { pathToFileURL } = await import("node:url");
206  const rel = "packages/database-engine/src/formula/index.ts";
207  const candidates: string[] = [join(process.cwd(), rel)];
208  // The importing file's directory (apps/api/{src,dist}/modules/ai is 5 levels
209  // under the repo root). Guarded: `__dirname` is CJS-only (undefined in ESM).
210  if (typeof __dirname !== "undefined") {
211    candidates.push(resolve(__dirname, "..", "..", "..", "..", "..", rel));
212  }
213  // Walk up from the process cwd — robust under vitest (cwd = a tests/ subdir),
214  // a monorepo tool, or any process launched from a subdirectory.
215  let dir = process.cwd();
216  for (let i = 0; i < 8; i++) {
217    candidates.push(join(dir, rel));
218    const parent = resolve(dir, "..");
219    if (parent === dir) break;
220    dir = parent;
221  }
222  for (const candidate of new Set(candidates)) {
223    if (existsSync(candidate)) {
224      try {
225        const mod = (await import(/* webpackIgnore: true */ pathToFileURL(candidate).href)) as DatabaseEngineModule;
226        if (typeof mod.validateFormula === "function" && typeof mod.evaluateFormula === "function") return mod;
227      } catch {
228        // broken source — keep trying
229      }
230    }
231  }
232  return undefined;
233}
234 
235@Injectable()
236export class DatabaseAiService {
237  constructor(
238    @Inject(PrismaService) private readonly prisma: PrismaClient,
239    @Inject(AI_GATEWAY) private readonly gateway: AiGateway,
240    @Inject(AiBudgetGuardService) private readonly guardService: AiBudgetGuardService,
241    @Inject(AiRateLimitService) private readonly rateLimit: AiRateLimitService,
242    @Inject(DatabaseService) private readonly databases: DatabaseService,
243    @Inject(DatabaseQueryService) private readonly records: DatabaseQueryService,
244    @Inject(PageService) private readonly pages: PageService,
245    @Inject(AuditService) private readonly audit: AuditService,
246  ) {}
247 
248  private get guard(): AiBudgetGuard {
249    return this.guardService.guard;
250  }
251 
252  /** Shared pre-stream gates: rate cap + budget (canonical JSON on failure). */
253  private async preStream(workspaceId: string, userId: string, req: Request): Promise<void> {
254    const rl = await this.rateLimit.check(userId);
255    if (!rl.allowed) {
256      throw new ApiError("RATE_LIMITED", 429, "AI call limit reached.", [
257        { reason: "ai_call_rate_limited", retryAfterSeconds: rl.retryAfterSeconds },
258      ]);
259    }
260    await primeWorkspaceCache(this.prisma, workspaceId);
261    try {
262      await this.guard.check(workspaceId, undefined, req);
263    } catch (err) {
264      throw mapAiError(err);
265    }
266  }
267 
268  private sseHeaders(res: Response): void {
269    res.status(200);
270    res.setHeader("Content-Type", "text/event-stream; charset=utf-8");
271    res.setHeader("Cache-Control", "no-cache, no-transform");
272    res.setHeader("Connection", "keep-alive");
273    res.setHeader("X-Accel-Buffering", "no");
274    res.flushHeaders?.();
275    if (typeof res.socket?.setNoDelay === "function") res.socket.setNoDelay(true);
276  }
277 
278  /**
279   * Run a tool-calling chat, collecting the first `toolCallDelta` + the final
280   * usage. Throws `no_tool_call` (422) when the model returns none. The
281   * `feedback` parameter is the retry-once seam (the previous error is fed
282   * back into the prompt).
283   */
284  private async chatToolCall(opts: {
285    workspaceId: string;
286    userId: string;
287    system: string;
288    user: string;
289    tool: typeof CREATE_DATABASE_TOOL | typeof INSERT_RECORDS_TOOL | typeof GENERATE_FORMULA_TOOL;
290    action: "generate" | "extract" | "formula";
291    feedback?: string;
292    signal?: AbortSignal;
293  }): Promise<{ args: unknown; toolId: string; usage: { in: number; out: number } }> {
294    const messages: Array<{ role: "system" | "user"; content: string }> = [
295      { role: "system", content: opts.system },
296      { role: "user", content: opts.feedback ? `${opts.user}\n\nPrevious attempt was INVALID: ${opts.feedback}\nCorrect it and call the tool again.` : opts.user },
297    ];
298    let args: unknown;
299    let toolId = "call";
300    let usage = { in: 0, out: 0 };
301    for await (const chunk of this.gateway.chat({
302      workspaceId: opts.workspaceId,
303      userId: opts.userId,
304      capability: "chat",
305      messages,
306      tools: [opts.tool],
307      toolChoice: { name: opts.tool.name },
308      metadata: { mode: "database", action: opts.action },
309      signal: opts.signal,
310    })) {
311      if (chunk.toolCallDelta) {
312        args = chunk.toolCallDelta.arguments;
313        toolId = chunk.toolCallDelta.id;
314      }
315      if (chunk.done && chunk.usage) usage = { in: chunk.usage.in, out: chunk.usage.out };
316    }
317    if (args === undefined) {
318      throw ApiError.unprocessable("The model returned no tool call.", [{ reason: "ai_invalid_structure", issue: "no tool call" }]);
319    }
320    return { args, toolId, usage };
321  }
322 
323  // ---------------------------------------------------------------- generate ---
324 
325  async generate(workspaceId: string, userId: string, input: AiGenerateDatabaseInput, req: Request, res: Response): Promise<void> {
326    // EDIT gate: creation goes through PageService.create (workspace EDIT).
327    await assertCan(this.prisma, { workspaceId, userId }, "EDIT");
328    await this.preStream(workspaceId, userId, req);
329 
330    const conversation = await this.prisma.aiConversation.create({
331      data: {
332        workspaceId,
333        userId,
334        title: DB_GEN_CONV_PREFIX,
335        context: { feature: "db.generate", parentId: input.parentId ?? null } as object,
336      },
337    });
338    await this.prisma.aiMessage.create({
339      data: { conversationId: conversation.id, role: "user", content: `generate: "${input.prompt.slice(0, 200)}"` },
340    });
341 
342    this.sseHeaders(res);
343    const ac = new AbortController();
344    let clientClosed = false;
345    const onClose = (): void => {
346      if (res.writableEnded) return;
347      clientClosed = true;
348      ac.abort();
349    };
350    res.on("close", onClose);
351    const timer = setTimeout(() => ac.abort(), TIMEOUT_MS);
352    const cleanup = (): void => {
353      clearTimeout(timer);
354      res.off("close", onClose);
355    };
356    const finish = (event: AiDatabaseSseEvent): void => {
357      cleanup();
358      if (!res.writableEnded) {
359        res.write(sseFrame(event));
360        res.end();
361      }
362    };
363 
364    try {
365      const userPrompt = renderGeneratePrompt(input.prompt);
366      let attempt = await this.chatToolCall({
367        workspaceId,
368        userId,
369        system: DB_GEN_SYSTEM_PROMPT,
370        user: userPrompt,
371        tool: CREATE_DATABASE_TOOL,
372        action: "generate",
373        signal: ac.signal,
374      });
375      let parsed = validateCreateDatabase(attempt.args);
376      if (!parsed.ok) {
377        attempt = await this.chatToolCall({
378          workspaceId,
379          userId,
380          system: DB_GEN_SYSTEM_PROMPT,
381          user: userPrompt,
382          tool: CREATE_DATABASE_TOOL,
383          action: "generate",
384          feedback: parsed.feedback,
385          signal: ac.signal,
386        });
387        parsed = validateCreateDatabase(attempt.args);
388      }
389 
390      if (!parsed.ok) {
391        // No partial write — nothing has been created yet.
392        const assistantId = await this.persistAssistant(conversation.id, this.gateway.resolve(workspaceId).chatModel, attempt.usage, Date.now(), {
393          feature: "db.generate",
394          error: "ai_invalid_structure",
395          issue: parsed.feedback,
396        });
397        if (!clientClosed && !res.writableEnded) res.write(sseFrame({ type: "tool_call", id: attempt.toolId, name: "createDatabase", arguments: attempt.args as Record<string, unknown> }));
398        if (!clientClosed && !res.writableEnded) {
399          res.write(sseFrame({
400            type: "error",
401            code: "UNPROCESSABLE_ENTITY",
402            message: "The AI returned an invalid database structure.",
403            reason: "ai_invalid_structure",
404          }));
405        }
406        finish({ type: "done", assistantMessageId: assistantId, reason: "error" });
407        return;
408      }
409 
410      const normalized = parsed.value;
411 
412      // ---- write through the normal services (permission/audit/realtime) ----
413      const page = await this.pages.create(workspaceId, userId, { title: normalized.title, parentId: input.parentId, type: "database" }, req);
414      // PageService.create returns a union (the template path is a different
415      // shape); we always pass type:"database" and never a templateId, so the
416      // `{ ..., databaseId }` member applies.
417      const databaseId = (page as { databaseId?: string }).databaseId!;
418 
419      const propertyIds: string[] = [];
420      for (const p of normalized.properties) {
421        const prop = await this.databases.createProperty(
422          databaseId,
423          userId,
424          {
425            name: p.name,
426            type: p.type,
427            ...(p.options ? { config: { options: p.options } } : { config: {} }),
428            ...(p.required !== undefined ? { required: p.required } : {}),
429          },
430          req,
431        );
432        propertyIds.push(prop.id);
433      }
434 
435      // property NAME → id (the seeded Title/Status + the new ones).
436      const allProps = await this.prisma.databaseProperty.findMany({
437        where: { databaseId, deletedAt: null },
438        select: { id: true, name: true },
439      });
440      const idByName = new Map(allProps.map((ap) => [ap.name.toLowerCase(), ap.id]));
441 
442      const recordIds: string[] = [];
443      for (const rec of normalized.sampleRecords) {
444        const title = typeof rec.Title === "string" ? rec.Title : undefined;
445        const properties: Record<string, unknown> = {};
446        for (const [name, value] of Object.entries(rec)) {
447          if (name === "Title") continue;
448          const pid = idByName.get(name.toLowerCase());
449          if (pid) properties[pid] = value;
450        }
451        const recDto = await this.records.createRecord(databaseId, userId, { title, properties, links: {} }, req);
452        recordIds.push(String((recDto as { id: string }).id));
453      }
454 
455      const assistantId = await this.persistAssistant(conversation.id, this.gateway.resolve(workspaceId).chatModel, attempt.usage, Date.now(), {
456        feature: "db.generate",
457        databaseId,
458        propertyIds,
459        recordIds,
460      });
461 
462      await this.audit.audit(req, req?.auth, {
463        action: "ai.database_generated",
464        resourceType: "database",
465        resourceId: databaseId,
466        workspaceId,
467        metadata: { count: recordIds.length, propertyCount: propertyIds.length },
468      });
469 
470      if (!clientClosed && !res.writableEnded) {
471        res.write(sseFrame({ type: "tool_call", id: attempt.toolId, name: "createDatabase", arguments: attempt.args as Record<string, unknown> }));
472        res.write(sseFrame({ type: "usage", tokensIn: attempt.usage.in, tokensOut: attempt.usage.out }));
473      }
474      finish({
475        type: "done",
476        assistantMessageId: assistantId,
477        reason: "ok",
478        databaseId,
479        propertyIds,
480        recordIds,
481        tokensIn: attempt.usage.in,
482        tokensOut: attempt.usage.out,
483      });
484    } catch (err) {
485      cleanup();
486      if (clientClosed || ac.signal.aborted) {
487        if (!clientClosed && !res.writableEnded) {
488          res.write(sseFrame({ type: "done", reason: "aborted" }));
489          res.end();
490        }
491        return;
492      }
493      const mapped = mapAiError(err);
494      if (!clientClosed && !res.writableEnded) {
495        res.write(sseFrame({ type: "error", code: mapped.code, message: mapped.message }));
496        res.write(sseFrame({ type: "done", reason: "error" }));
497        res.end();
498      }
499      logger.info(`ai database generate stream error workspace=${workspaceId} code=${mapped.code}`);
500    }
501  }
502 
503  // ----------------------------------------------------------------- extract ---
504 
505  async extract(workspaceId: string, userId: string, input: AiExtractInput, req: Request, res: Response): Promise<void> {
506    const db = await this.prisma.database.findFirst({
507      where: { id: input.databaseId, workspaceId, isArchived: false },
508      select: { id: true, title: true },
509    });
510    if (!db) throw ApiError.notFound("Database not found.");
511    // EDIT on the database (record creation is an EDIT action).
512    await assertCan(this.prisma, { workspaceId, userId, resource: { databaseId: db.id } }, "EDIT");
513    await this.preStream(workspaceId, userId, req);
514 
515    const properties = await this.loadProps(db.id);
516 
517    const conversation = await this.prisma.aiConversation.create({
518      data: {
519        workspaceId,
520        userId,
521        title: DB_EXTRACT_CONV_PREFIX,
522        context: { feature: "db.extract", databaseId: db.id } as object,
523      },
524    });
525    await this.prisma.aiMessage.create({
526      data: { conversationId: conversation.id, role: "user", content: `extract: ${input.text.length} chars → ${db.title}` },
527    });
528 
529    this.sseHeaders(res);
530    const ac = new AbortController();
531    let clientClosed = false;
532    const onClose = (): void => {
533      if (res.writableEnded) return;
534      clientClosed = true;
535      ac.abort();
536    };
537    res.on("close", onClose);
538    const timer = setTimeout(() => ac.abort(), TIMEOUT_MS);
539    const cleanup = (): void => {
540      clearTimeout(timer);
541      res.off("close", onClose);
542    };
543    const finish = (event: AiDatabaseSseEvent): void => {
544      cleanup();
545      if (!res.writableEnded) {
546        res.write(sseFrame(event));
547        res.end();
548      }
549    };
550 
551    try {
552      const prompt = renderExtractPrompt({ databaseTitle: db.title, properties, text: input.text });
553      let attempt = await this.chatToolCall({ workspaceId, userId, system: DB_EXTRACT_SYSTEM_PROMPT, user: prompt, tool: INSERT_RECORDS_TOOL, action: "extract", signal: ac.signal });
554      let parsed = validateInsertRecords(attempt.args, properties);
555      if (!parsed.ok) {
556        attempt = await this.chatToolCall({ workspaceId, userId, system: DB_EXTRACT_SYSTEM_PROMPT, user: prompt, tool: INSERT_RECORDS_TOOL, action: "extract", feedback: parsed.feedback, signal: ac.signal });
557        parsed = validateInsertRecords(attempt.args, properties);
558      }
559 
560      if (!parsed.ok) {
561        const assistantId = await this.persistAssistant(conversation.id, this.gateway.resolve(workspaceId).chatModel, attempt.usage, Date.now(), {
562          feature: "db.extract",
563          databaseId: db.id,
564          error: "ai_invalid_structure",
565          issue: parsed.feedback,
566        });
567        if (!clientClosed && !res.writableEnded) res.write(sseFrame({ type: "tool_call", id: attempt.toolId, name: "insertRecords", arguments: attempt.args as Record<string, unknown> }));
568        if (!clientClosed && !res.writableEnded) {
569          res.write(sseFrame({
570            type: "error",
571            code: "UNPROCESSABLE_ENTITY",
572            message: "The AI returned an invalid record structure.",
573            reason: "ai_invalid_structure",
574          }));
575        }
576        finish({ type: "done", assistantMessageId: assistantId, reason: "error" });
577        return;
578      }
579 
580      const normalized = parsed.value;
581      const recordIds: string[] = [];
582      for (const rec of normalized.records) {
583        const title = typeof rec.properties.__title__ === "string" ? (rec.properties.__title__ as string) : undefined;
584        const { __title__, ...properties } = rec.properties as Record<string, unknown> & { __title__?: string };
585        const recDto = await this.records.createRecord(db.id, userId, { title, properties, links: {} }, req);
586        recordIds.push(String((recDto as { id: string }).id));
587      }
588 
589      const assistantId = await this.persistAssistant(conversation.id, this.gateway.resolve(workspaceId).chatModel, attempt.usage, Date.now(), {
590        feature: "db.extract",
591        databaseId: db.id,
592        recordIds,
593        ignoredColumns: normalized.ignoredColumns,
594      });
595 
596      await this.audit.audit(req, req?.auth, {
597        action: "ai.records_extracted",
598        resourceType: "database",
599        resourceId: db.id,
600        workspaceId,
601        metadata: { count: recordIds.length, ignoredColumns: normalized.ignoredColumns },
602      });
603 
604      if (!clientClosed && !res.writableEnded) {
605        res.write(sseFrame({ type: "tool_call", id: attempt.toolId, name: "insertRecords", arguments: attempt.args as Record<string, unknown> }));
606        res.write(sseFrame({ type: "usage", tokensIn: attempt.usage.in, tokensOut: attempt.usage.out }));
607      }
608      finish({
609        type: "done",
610        assistantMessageId: assistantId,
611        reason: "ok",
612        databaseId: db.id,
613        recordIds,
614        ignoredColumns: normalized.ignoredColumns,
615        tokensIn: attempt.usage.in,
616        tokensOut: attempt.usage.out,
617      });
618    } catch (err) {
619      cleanup();
620      if (clientClosed || ac.signal.aborted) {
621        if (!clientClosed && !res.writableEnded) {
622          res.write(sseFrame({ type: "done", reason: "aborted" }));
623          res.end();
624        }
625        return;
626      }
627      const mapped = mapAiError(err);
628      if (!clientClosed && !res.writableEnded) {
629        res.write(sseFrame({ type: "error", code: mapped.code, message: mapped.message }));
630        res.write(sseFrame({ type: "done", reason: "error" }));
631        res.end();
632      }
633      logger.info(`ai database extract stream error workspace=${workspaceId} code=${mapped.code}`);
634    }
635  }
636 
637  // ----------------------------------------------------------------- preview ---
638 
639  /** Deterministic column→property mapping (no LLM, no write, no budget). */
640  async preview(workspaceId: string, userId: string, input: AiExtractPreviewInput) {
641    const db = await this.prisma.database.findFirst({
642      where: { id: input.databaseId, workspaceId, isArchived: false },
643      select: { id: true },
644    });
645    if (!db) throw ApiError.notFound("Database not found.");
646    await assertCan(this.prisma, { workspaceId, userId, resource: { databaseId: db.id } }, "EDIT");
647    const properties = await this.loadProps(db.id);
648    const mapping = previewExtractMapping(input.text, properties);
649    return { databaseId: db.id, ...mapping };
650  }
651 
652  // ----------------------------------------------------------------- formula ---
653 
654  async formula(
655    workspaceId: string,
656    userId: string,
657    input: AiGenerateFormulaInput,
658    req: Request,
659  ): Promise<{
660    formulaId: string;
661    propertyId: string;
662    expression: string;
663    returnType: string;
664    explanation?: string;
665    samples: Array<{ recordId: string; value: unknown; error?: string }>;
666    tokensIn: number;
667    tokensOut: number;
668  }> {
669    const db = await this.prisma.database.findFirst({
670      where: { id: input.databaseId, workspaceId, isArchived: false },
671      select: { id: true },
672    });
673    if (!db) throw ApiError.notFound("Database not found.");
674    await assertCan(this.prisma, { workspaceId, userId, resource: { databaseId: db.id } }, "EDIT");
675    await this.preStream(workspaceId, userId, req);
676 
677    const engine = await loadDatabaseEngine();
678    if (!engine) {
679      throw new ApiError("SERVICE_UNAVAILABLE", 503, "The formula engine is unavailable.", [
680        { reason: "formula_engine_unavailable" },
681      ]);
682    }
683 
684    const properties = await this.loadProps(db.id);
685    const propNames = properties.map((p) => ({ name: p.name, type: p.type }));
686    const propertyTypes = new Map(properties.map((p) => [p.name, lowerPropertyType(p.type)]));
687    const sampleRows = await this.loadSampleRows(db.id, input.sampleRowIds);
688 
689    const conversation = await this.prisma.aiConversation.create({
690      data: {
691        workspaceId,
692        userId,
693        title: DB_FORMULA_CONV_PREFIX,
694        context: { feature: "db.formula", databaseId: db.id } as object,
695      },
696    });
697    await this.prisma.aiMessage.create({
698      data: { conversationId: conversation.id, role: "user", content: `formula: "${input.description.slice(0, 200)}"` },
699    });
700 
701    const userPrompt = renderFormulaPrompt({ properties: propNames, description: input.description });
702    let attempt = await this.chatToolCall({ workspaceId, userId, system: DB_FORMULA_SYSTEM_PROMPT, user: userPrompt, tool: GENERATE_FORMULA_TOOL, action: "formula" });
703    let expression = String((attempt.args as { expression?: unknown })?.expression ?? "");
704    let explanation = typeof (attempt.args as { explanation?: unknown })?.explanation === "string" ? ((attempt.args as { explanation?: string }).explanation as string) : undefined;
705 
706    // Validate with the safe engine BEFORE saving (parse + type-check).
707    // A record's formula values = its stored properties + its TITLE (the
708    // title lives on the page, not a RecordProperty — but prop("Title") must
709    // resolve). Missing properties are null (the engine normalizes null).
710    const formulaPropsFor = (row: { title: string | null; properties: Array<{ property: { id: string; name: string; type: string; config: unknown }; value: unknown }> }): Record<string, unknown> => {
711      const out = recordToFormulaProps(row.properties);
712      out.Title = row.title ?? "";
713      return out;
714    };
715    const sampleValues: Record<string, unknown> = {};
716    for (const p of properties) sampleValues[p.name] = null;
717    if (sampleRows.length > 0) Object.assign(sampleValues, formulaPropsFor(sampleRows[0]!));
718    const typeMap = Object.fromEntries(propertyTypes);
719 
720    let validation: ValidatedFormulaLike | null = null;
721    let parseError: string | undefined;
722    const validate = (expr: string): boolean => {
723      try {
724        validation = engine.validateFormula(expr, sampleValues, typeMap);
725        return true;
726      } catch (err) {
727        parseError = err instanceof engine.FormulaError ? `${err.code}: ${err.message}` : (err as Error).message;
728        return false;
729      }
730    };
731 
732    if (!validate(expression)) {
733      // Retry once with the parse error as feedback.
734      attempt = await this.chatToolCall({ workspaceId, userId, system: DB_FORMULA_SYSTEM_PROMPT, user: userPrompt, tool: GENERATE_FORMULA_TOOL, action: "formula", feedback: parseError ?? "the expression is invalid" });
735      expression = String((attempt.args as { expression?: unknown })?.expression ?? "");
736      explanation = typeof (attempt.args as { explanation?: unknown })?.explanation === "string" ? ((attempt.args as { explanation?: string }).explanation as string) : undefined;
737      if (!validate(expression)) {
738        // No Formula row written.
739        await this.persistAssistant(conversation.id, this.gateway.resolve(workspaceId).chatModel, attempt.usage, Date.now(), {
740          feature: "db.formula",
741          databaseId: db.id,
742          error: "formula_invalid",
743          parseError,
744        });
745        throw ApiError.unprocessable("The AI generated a formula that failed validation.", [
746          { reason: "formula_invalid", parseError: parseError ?? null },
747        ]);
748      }
749    }
750 
751    // Evaluate on the sample rows (read-only; the engine's guards apply).
752    // The evaluator returns a typed EValue `{v, t}` — `v` is the raw value.
753    const samples = sampleRows.map((row) => {
754      try {
755        const { value } = engine.evaluateFormula(validation!.ast, {
756          now: Date.now(),
757          properties: formulaPropsFor(row),
758          propertyTypes: typeMap,
759        });
760        return { recordId: row.id, value: value.v };
761      } catch (err) {
762        return { recordId: row.id, value: null, error: err instanceof engine.FormulaError ? err.message : (err as Error).message };
763      }
764    });
765 
766    // ---- save: formula property + Formula row through the normal path
767    // (createFormula re-checks EDIT, audits `database.formula_created`, and
768    // publishes `database.updated` realtime). The engine-validated AST +
769    // returnType are stored verbatim (spec: the expression is engine-validated
770    // BEFORE saving).
771    const name = input.name ?? "Formula";
772    const formula = await this.databases.createFormula(
773      db.id,
774      userId,
775      { name, expression, ast: validation!.ast as object, returnType: validation!.returnType },
776      req,
777    );
778    const propertyId = formula.propertyId;
779 
780    await this.persistAssistant(conversation.id, this.gateway.resolve(workspaceId).chatModel, attempt.usage, Date.now(), {
781      feature: "db.formula",
782      databaseId: db.id,
783      formulaId: formula.id,
784      propertyId,
785      expression,
786      samples,
787    });
788 
789    await this.audit.audit(req, req?.auth, {
790      action: "ai.formula_generated",
791      resourceType: "formula",
792      resourceId: formula.id,
793      workspaceId,
794      metadata: { databaseId: db.id, expressionHash: hashExpression(expression) },
795    });
796 
797    return {
798      formulaId: formula.id,
799      propertyId: propertyId!,
800      expression,
801      returnType: validation!.returnType,
802      ...(explanation ? { explanation } : {}),
803      samples,
804      tokensIn: attempt.usage.in,
805      tokensOut: attempt.usage.out,
806    };
807  }
808 
809  // ------------------------------------------------------------------ helpers ---
810 
811  private async loadProps(databaseId: string): Promise<DbPropMeta[]> {
812    const rows = await this.prisma.databaseProperty.findMany({
813      where: { databaseId, deletedAt: null },
814      orderBy: { position: "asc" },
815      select: { id: true, name: true, type: true, isTitle: true, config: true },
816    });
817    return rows.map((r) => ({ id: r.id, name: r.name, type: r.type, isTitle: r.isTitle, config: r.config }));
818  }
819 
820  private async loadSampleRows(
821    databaseId: string,
822    sampleRowIds?: string[],
823  ): Promise<Array<{ id: string; title: string | null; properties: Array<{ property: { id: string; name: string; type: string; config: unknown }; value: unknown }> }>> {
824    const where =
825      sampleRowIds && sampleRowIds.length > 0
826        ? { id: { in: sampleRowIds.slice(0, DB_FORMULA_MAX_SAMPLE_ROWS) }, databaseId, archivedAt: null }
827        : { databaseId, archivedAt: null };
828    const rows = await this.prisma.databaseRecord.findMany({
829      where,
830      include: {
831        properties: { select: { property: { select: { id: true, name: true, type: true, config: true } }, value: true } },
832        // The record IS a page — the TITLE property lives on the page, and a
833        // formula may reference it (prop("Title")). Carried separately because
834        // it is not a RecordProperty row.
835        page: { select: { title: true } },
836      },
837      orderBy: { position: "asc" },
838      take: sampleRowIds ? undefined : DB_FORMULA_MAX_SAMPLE_ROWS,
839    });
840    // The record's title lives on its page (the record IS a page) — the
841    // Prisma row carries it at `page.title`, but the formula values need a
842    // flat `title` (a formula may reference prop("Title")).
843    return rows.map((r) => ({
844      id: r.id,
845      title: r.page?.title ?? null,
846      properties: r.properties,
847    }));
848  }
849 
850  private async persistAssistant(
851    conversationId: string,
852    model: string,
853    usage: { in: number; out: number },
854    _now: number,
855    metadata: Record<string, unknown>,
856  ): Promise<string | undefined> {
857    try {
858      const row = await this.prisma.aiMessage.create({
859        data: {
860          conversationId,
861          role: "assistant",
862          content: metadata.error ? `error: ${String(metadata.error)}` : "ok",
863          model,
864          ...(usage.in > 0 ? { tokensIn: usage.in } : {}),
865          ...(usage.out > 0 ? { tokensOut: usage.out } : {}),
866          metadata: metadata as object,
867        },
868      });
869      return row.id;
870    } catch (err) {
871      logger.error(`ai database assistant persist failed: ${(err as Error).message}`);
872      return undefined;
873    }
874  }
875}
876