spb/chat-spboucher-ai

1import "server-only";2import { sql } from "../db";3import { adapterFor, estimateCost, route, type RoutingMode, type AIModel, type Part, type UnifiedMessage, type Usage, type ChatSettings } from "../ai";4import { listModels, requireChatModel } from "../ai/registry";5import { readAttachment } from "../storage";6import { recordUsage, assertWithinBudget } from "../usage";7import { TOOLS, toolDefs, toolByName, type ToolContext } from "../tools";8import { buildSystemPrompt } from "./system";9import { getConversation, getActiveMessages, insertMessage, updateMessage, type MessageRow } from "./store";10import { newId } from "../ids";11import { maybeTitle, extractMemories, detectArtifacts, embedMessages } from "./post";1213export type Emit = (event: string, data: unknown) => void;1415export interface RunOptions {16  conversationId: string;17  modelId?: string;                 // requested (manual) model18  routingMode?: RoutingMode;19  settings?: ChatSettings;20  toolsEnabled?: boolean;21  toolNames?: string[];22  /** when regenerating, the assistant message to replace (its parent stays) */23  parentMessageId: string;          // the user message this reply answers24  signal?: AbortSignal;25  emit: Emit;26  approvedToolCallIds?: string[];27}2829const MAX_TOOL_ROUNDS = 8;3031export async function runAssistant(o: RunOptions): Promise<{ messageId: string; status: string }> {32  const conv = await getConversation(o.conversationId);33  if (!conv) throw Object.assign(new Error("Conversation introuvable"), { status: 404 });34  const budget = await assertWithinBudget();35  if (budget.warning) o.emit("warning", { message: budget.warning });3637  const history = await getActiveMessages(o.conversationId);38  const upTo = history.findIndex(m => m.id === o.parentMessageId);39  const thread = upTo >= 0 ? history.slice(0, upTo + 1) : history;40  const lastUser = [...thread].reverse().find(m => m.role === "user");41  const queryText = lastUser?.text ?? "";42  const hasImages = thread.some(m => m.parts.some(p => p.type === "image"));4344  // ---- routing45  const mode: RoutingMode = o.routingMode ?? (o.modelId ? "manual" : "auto");46  let model: AIModel;47  let routing: Record<string, unknown>;48  if (mode === "manual") { model = await requireChatModel(o.modelId ?? conv.model_id ?? ""); routing = { mode, selected: model.id, reason: "Modèle choisi explicitement." }; }49  else {50    const all = await listModels({ onlyChat: true });51    const approxTokens = Math.round(thread.reduce((s, m) => s + m.text.length, 0) / 3.5);52    const d = route(mode, all, o.modelId, queryText, { hasImages, approxTokens, wantsTools: o.toolsEnabled });53    model = await requireChatModel(d.selected); routing = { ...d };54  }55  const toolsOn = (o.toolsEnabled ?? true) && model.capabilities.tools === true;56  const projPerm = (conv.project_id ? ((await sql<{ tool_permissions: Record<string, boolean> }[]>`SELECT tool_permissions FROM projects WHERE id = ${conv.project_id}`)[0]?.tool_permissions ?? {}) : {}) as Record<string, boolean>;57  const allowedNames = TOOLS.map(t => t.name).filter(n => projPerm[n] !== false && (!o.toolNames || o.toolNames.includes(n)));58  const tools = toolsOn ? toolDefs(allowedNames) : [];5960  // ---- attachments summary61  const atts = await sql<{ id: string; filename: string; mime: string; page_count: number | null; extracted_text: string | null }[]>`SELECT id, filename, mime, page_count, extracted_text FROM attachments WHERE conversation_id = ${o.conversationId} AND kind = 'upload' AND mime NOT LIKE 'image/%' ORDER BY created_at`;62  const attachmentsSummary = atts.length ? atts.map(a => `- ${a.filename} (${a.mime}${a.page_count ? `, ${a.page_count} p.` : ""}${a.extracted_text ? `, ${a.extracted_text.length.toLocaleString("fr-CA")} car.` : ""})`).join("\n") : undefined;6364  const { system, memoriesUsed, project } = await buildSystemPrompt({ modelId: model.id, projectId: conv.project_id, query: queryText, attachmentsSummary, conversationSettings: conv.settings, toolsEnabled: toolsOn });65  const settings: ChatSettings = { ...(conv.settings?.chat as ChatSettings | undefined), ...o.settings };6667  // ---- assistant message row (streaming)68  const assistant = await insertMessage({ conversationId: o.conversationId, parentId: o.parentMessageId, role: "assistant", parts: [], modelId: model.id, provider: model.provider, settings: settings as Record<string, unknown>, routing, status: "streaming" });69  o.emit("start", { messageId: assistant.id, model: model.id, provider: model.provider, routing, memoriesUsed: memoriesUsed.map(m => m.id) });7071  const parts: Part[] = [];72  const unified: UnifiedMessage[] = thread.map(m => ({ role: m.role, parts: inlineShortDocs(m.parts, atts) }));73  const loadAttachment = async (id: string) => readAttachment(id);74  const t0 = Date.now(); let ttft: number | undefined; let totalUsage: Usage = { input: 0, output: 0, cached: 0, reasoning: 0 };75  let status = "complete"; let error: string | null = null; let lastFlush = 0;76  const flush = async (force = false) => { if (force || Date.now() - lastFlush > 700) { lastFlush = Date.now(); await updateMessage(assistant.id, { parts }); } };77  const ctx: ToolContext = { conversationId: o.conversationId, projectId: conv.project_id, messageId: assistant.id, signal: o.signal, allow: new Set(o.approvedToolCallIds ?? []) };7879  try {80    for (let round = 0; round <= MAX_TOOL_ROUNDS; round++) {81      const adapter = adapterFor(model.provider);82      let text = ""; let reasoning = ""; const calls: { id: string; name: string; input: unknown }[] = []; let stop = "end_turn"; let stopDetails: unknown;83      let textPart: Extract<Part, { type: "text" }> | null = null; let reasoningPart: Extract<Part, { type: "reasoning" }> | null = null;84      for await (const ev of adapter.stream({ model, system, messages: [...unified, ...(parts.length ? [{ role: "assistant" as const, parts }] : [])], tools, settings, signal: o.signal, loadAttachment })) {85        if (ev.type === "text") {86          if (ttft === undefined) ttft = Date.now() - t0;87          text += ev.delta;88          if (!textPart) { textPart = { type: "text", text: "" }; parts.push(textPart); }89          textPart.text = text; o.emit("text", { delta: ev.delta }); await flush();90        } else if (ev.type === "reasoning") {91          reasoning += ev.delta;92          if (!reasoningPart) { reasoningPart = { type: "reasoning", text: "", summarized: true }; parts.splice(textPart ? parts.indexOf(textPart) : parts.length, 0, reasoningPart); }93          reasoningPart.text = reasoning; o.emit("reasoning", { delta: ev.delta });94        } else if (ev.type === "tool_call") { calls.push(ev); }95        else if (ev.type === "usage") { totalUsage = { input: totalUsage.input + ev.usage.input, output: totalUsage.output + ev.usage.output, cached: (totalUsage.cached ?? 0) + (ev.usage.cached ?? 0), reasoning: (totalUsage.reasoning ?? 0) + (ev.usage.reasoning ?? 0) }; }96        else if (ev.type === "stop") { stop = ev.reason; stopDetails = ev.details; }97        else if (ev.type === "meta") { o.emit("meta", ev.data); if (ev.data.servedBy) routing.servedBy = ev.data.servedBy; }98        else if (ev.type === "error") { throw Object.assign(new Error(ev.message), { status: ev.status }); }99      }100      if (stop === "cancelled") { status = "cancelled"; break; }101      if (stop === "refusal") { o.emit("warning", { message: `Le modèle a refusé de répondre (${JSON.stringify(stopDetails ?? {})}).` }); }102      if (!calls.length) break;103      // ---- execute tools (parallel), append tool_call parts with results104      const pending: Part[] = [];105      for (const c of calls) {106        const tool = toolByName(c.name);107        const part: Extract<Part, { type: "tool_call" }> = { type: "tool_call", id: c.id, name: c.name, input: c.input, status: "running" };108        parts.push(part); pending.push(part);109        o.emit("tool_call", { id: c.id, name: c.name, input: c.input, status: tool ? (tool.permission === "approval" && !ctx.allow.has(c.id) ? "pending_approval" : "running") : "error" });110        if (!tool) { part.status = "error"; part.error = `Unknown tool ${c.name}`; continue; }111        if (tool.permission === "approval" && !ctx.allow.has(c.id)) { part.status = "pending_approval"; continue; }112      }113      await flush(true);114      if (pending.some(p => p.type === "tool_call" && p.status === "pending_approval")) { status = "awaiting_approval"; break; }115      await Promise.all(pending.map(async p => {116        if (p.type !== "tool_call" || p.status !== "running") return;117        const tool = toolByName(p.name)!; const ts = Date.now(); const execId = newId("tx");118        try {119          const out = await tool.run((p.input ?? {}) as Record<string, unknown>, ctx);120          p.output = out; p.status = "ok"; p.durationMs = Date.now() - ts;121          if (p.name === "generate_image" && out && typeof out === "object" && "attachmentId" in out) { const gi: Part = { type: "generated_image", attachmentId: String((out as { attachmentId: string }).attachmentId), prompt: String((p.input as { prompt?: string })?.prompt ?? ""), model: String((out as { model?: string }).model ?? "") }; parts.push(gi); o.emit("part", gi); }122        } catch (e) { p.status = "error"; p.error = String((e as Error).message ?? e).slice(0, 500); p.durationMs = Date.now() - ts; }123        await sql`INSERT INTO tool_executions (id, conversation_id, message_id, tool, input, output, status, duration_ms) VALUES (${execId}, ${o.conversationId}, ${assistant.id}, ${p.name}, ${sql.json(p.input as never)}, ${sql.json((p.status === "ok" ? p.output : { error: p.error }) as never)}, ${p.status}, ${p.durationMs ?? null})`;124        o.emit("tool_result", { id: p.id, status: p.status, output: p.output, error: p.error, durationMs: p.durationMs });125      }));126      await flush(true);127      if (round === MAX_TOOL_ROUNDS) { o.emit("warning", { message: "Nombre maximal d'appels d'outils atteint." }); }128    }129  } catch (e) {130    const err = e as Error & { status?: number };131    if (o.signal?.aborted) status = "cancelled"; else { status = "error"; error = err.message; o.emit("error", { message: err.message, status: err.status }); }132  }133134  const latency = Date.now() - t0;135  const cost = estimateCost(model, totalUsage);136  await updateMessage(assistant.id, { parts, status, error, usage: totalUsage as unknown as Record<string, number>, costUsd: cost, latencyMs: latency, ttftMs: ttft ?? null, routing });137  await recordUsage({ provider: model.provider, modelId: model.id, kind: conv.mode === "research" ? "research" : "chat", conversationId: o.conversationId, projectId: conv.project_id, messageId: assistant.id, usage: totalUsage, costEstimated: cost, latencyMs: latency, ttftMs: ttft, status: status === "error" ? "error" : status === "cancelled" ? "cancelled" : "ok", error });138  o.emit("usage", { usage: totalUsage, costUsd: cost, latencyMs: latency, ttftMs: ttft ?? null, model: model.id });139140  // ---- post-processing (non-blocking for the client, but awaited so the SSE stays open briefly)141  const artifacts = await detectArtifacts(assistant.id, o.conversationId, conv.project_id, parts);142  if (artifacts.length) o.emit("artifacts", artifacts);143  void Promise.allSettled([144    maybeTitle(conv, thread, parts).then(t => { if (t) o.emit("title", { title: t }); }),145    lastUser && status === "complete" ? extractMemories(conv, lastUser as MessageRow, parts, project?.id ?? null) : Promise.resolve(),146    embedMessages([lastUser?.id, assistant.id].filter((x): x is string => !!x)),147  ]).then(() => o.emit("done", { messageId: assistant.id, status })).catch(() => o.emit("done", { messageId: assistant.id, status }));148  return { messageId: assistant.id, status };149}150151/** Inline the extracted text of short documents directly in the user parts (long ones go through search_documents). */152function inlineShortDocs(parts: Part[], atts: { id: string; filename: string; extracted_text: string | null }[]): Part[] {153  return parts.flatMap((p): Part[] => {154    if (p.type !== "file") return [p];155    const a = atts.find(x => x.id === p.attachmentId);156    if (a?.extracted_text && a.extracted_text.length <= 60_000 && p.mime !== "application/pdf") return [{ type: "text" as const, text: `<file name="${a.filename}">\n${a.extracted_text}\n</file>` }];157    return [p];158  });159}160