import "server-only"; import { sql } from "../db"; import { newId } from "../ids"; import { adapterFor, estimateCost, type AIModel } from "../ai"; import { requireChatModel } from "../ai/registry"; import { recordUsage, assertWithinBudget } from "../usage"; import { webSearch, fetchPage } from "../tools/web"; import { utilityModel } from "../chat/post"; import { createConversation, insertMessage, updateMessage } from "../chat/store"; export type Emit = (event: string, data: unknown) => void; interface Source { n: number; url: string; title: string; snippet: string; publishedAt?: string; text?: string; fetched: boolean; evidence?: { claim: string; quote: string; confidence: "high" | "medium" | "low" }[] } async function complete(model: AIModel, prompt: string, kind: string, runId: string, maxTokens = 6000, effort?: "low" | "medium" | "high" | "xhigh" | "max", signal?: AbortSignal): Promise<{ text: string; cost: number }> { let text = ""; let usage = { input: 0, output: 0, cached: 0, reasoning: 0 }; const t0 = Date.now(); for await (const ev of adapterFor(model.provider).stream({ model, messages: [{ role: "user", parts: [{ type: "text", text: prompt }] }], settings: { maxOutputTokens: maxTokens, effort, reasoningSummary: false }, signal, loadAttachment: async () => null })) { if (ev.type === "text") text += ev.delta; else if (ev.type === "usage") usage = { input: usage.input + ev.usage.input, output: usage.output + ev.usage.output, cached: usage.cached + (ev.usage.cached ?? 0), reasoning: usage.reasoning + (ev.usage.reasoning ?? 0) }; else if (ev.type === "error") throw new Error(ev.message); } const cost = estimateCost(model, usage) ?? 0; await recordUsage({ provider: model.provider, modelId: model.id, kind, usage, costEstimated: cost, latencyMs: Date.now() - t0, meta: { researchId: runId } }); return { text, cost }; } const parseJson = (s: string, fallback: T): T => { try { return JSON.parse(s.replace(/```(?:json)?/g, "").trim()) as T; } catch { const m = s.match(/[\[{][\s\S]*[\]}]/); if (m) { try { return JSON.parse(m[0]) as T; } catch { /* */ } } return fallback; } }; export async function runResearch(o: { question: string; modelId: string; projectId?: string | null; depth?: "quick" | "standard" | "deep"; emit: Emit; signal?: AbortSignal }): Promise<{ id: string; conversationId: string }> { await assertWithinBudget(); const model = await requireChatModel(o.modelId); const extractor = (await utilityModel()) ?? model; const depth = o.depth ?? "standard"; const nQueries = depth === "quick" ? 3 : depth === "deep" ? 8 : 5; const nFetch = depth === "quick" ? 4 : depth === "deep" ? 12 : 7; const id = newId("rs"); const conv = await createConversation({ projectId: o.projectId ?? null, modelId: model.id, title: o.question.slice(0, 80), mode: "research" }); const userMsg = await insertMessage({ conversationId: conv.id, role: "user", parts: [{ type: "text", text: o.question }] }); const assistant = await insertMessage({ conversationId: conv.id, parentId: userMsg.id, role: "assistant", parts: [], modelId: model.id, provider: model.provider, status: "streaming" }); await sql`INSERT INTO research_runs (id, conversation_id, project_id, question, model_id, status) VALUES (${id}, ${conv.id}, ${o.projectId ?? null}, ${o.question}, ${model.id}, 'planning')`; let cost = 0; const steps: { step: string; at: string; detail?: unknown }[] = []; const step = async (name: string, status: string, detail?: unknown) => { steps.push({ step: name, at: new Date().toISOString(), detail }); o.emit("step", { step: name, status, detail }); await sql`UPDATE research_runs SET status = ${status}, steps = ${sql.json(steps as never)}, cost_usd = ${cost} WHERE id = ${id}`; }; o.emit("start", { id, conversationId: conv.id, model: model.id, extractor: extractor.id, depth }); try { // 1. plan const planRaw = await complete(model, `You are planning a research investigation. Question: """${o.question}"""\nToday: ${new Date().toISOString().slice(0, 10)}.\nReturn ONLY JSON: {"subquestions": [3-6 precise sub-questions], "queries": [${nQueries} diverse web search queries in the most useful language(s), mixing broad and specific], "evidence_needed": [what kinds of sources would settle it]}.`, "research", id, 2000, "low", o.signal); cost += planRaw.cost; const plan = parseJson<{ subquestions: string[]; queries: string[]; evidence_needed: string[] }>(planRaw.text, { subquestions: [o.question], queries: [o.question], evidence_needed: [] }); plan.queries = (plan.queries ?? [o.question]).slice(0, nQueries); await sql`UPDATE research_runs SET plan = ${sql.json(plan as never)} WHERE id = ${id}`; await step("plan", "searching", plan); // 2. discovery const seen = new Map(); const searches = await Promise.all(plan.queries.map(q => webSearch(q, { maxResults: 8 }))); for (const s of searches) if ("results" in s) for (const r of s.results) if (!seen.has(r.url) && !/\.(pdf|zip)$/i.test(r.url)) seen.set(r.url, { n: seen.size + 1, url: r.url, title: r.title, snippet: r.snippet, publishedAt: r.publishedAt, fetched: false }); const sources = [...seen.values()]; if (!sources.length) throw new Error(searches.map(s => "error" in s ? s.error : "").filter(Boolean)[0] ?? "Aucune source trouvée"); await step("discovery", "reading", { queries: plan.queries, found: sources.length }); // 3. retrieval (top N by snippet relevance order as returned) const toFetch = sources.slice(0, nFetch); await Promise.all(toFetch.map(async s => { const r = await fetchPage(s.url); if ("text" in r && r.text.length > 300) { s.text = r.text.slice(0, 12_000); s.fetched = true; if (r.title && !s.title) s.title = r.title; } o.emit("source", { n: s.n, url: s.url, title: s.title, fetched: s.fetched }); })); await step("retrieval", "reading", { fetched: toFetch.filter(s => s.fetched).length, of: toFetch.length }); // 4. evidence extraction (cheap model, parallel) await Promise.all(toFetch.filter(s => s.fetched).map(async s => { const r = await complete(extractor, `Extract evidence relevant to the research question from this source. Question: """${o.question}"""\nSub-questions: ${plan.subquestions.join(" | ")}\nSource [${s.n}] ${s.title} — ${s.url}\n"""\n${s.text}\n"""\nReturn ONLY JSON: [{"claim": "one factual statement supported by the source", "quote": "short verbatim excerpt (<= 30 words)", "confidence": "high|medium|low"}] (max 6 items, empty array if nothing relevant).`, "research", id, 1500, "low", o.signal); cost += r.cost; s.evidence = parseJson(r.text, []); o.emit("evidence", { n: s.n, count: s.evidence?.length ?? 0 }); })); await step("extraction", "synthesizing", { claims: toFetch.reduce((a, s) => a + (s.evidence?.length ?? 0), 0) }); // 5+6. cross-validation & synthesis (selected model, streamed) const dossier = toFetch.filter(s => s.fetched).map(s => `[${s.n}] ${s.title} — ${s.url}${s.publishedAt ? ` (${s.publishedAt})` : ""}\n${(s.evidence ?? []).map(e => ` • (${e.confidence}) ${e.claim} — « ${e.quote} »`).join("\n") || " • (no structured evidence extracted; snippet: " + s.snippet.slice(0, 300) + ")"}`).join("\n\n"); const unread = sources.slice(nFetch).map(s => `[${s.n}] ${s.title} — ${s.url} (not read; snippet: ${s.snippet.slice(0, 200)})`).join("\n"); const synthPrompt = `You are writing a rigorous research report in the language of the question.\nQuestion: """${o.question}"""\nSub-questions: ${plan.subquestions.map((q, i) => `${i + 1}. ${q}`).join(" ")}\n\nEVIDENCE DOSSIER (numbered sources; cite them as [n]):\n${dossier}\n\nADDITIONAL SOURCES FOUND BUT NOT READ:\n${unread || "(none)"}\n\nWrite the report in Markdown with: a 3-5 sentence executive summary; one section per sub-question; a "Cross-validation" section listing where sources agree, where they conflict, and what remains unverified; a "Limitations" section; and a final "Sources" list ([n] title — URL). Rules: every factual sentence that relies on a source carries a [n] citation; clearly label interpretation ("Interprétation :" / "Interpretation:") and speculation; never cite a source for something it does not say; do not invent sources or numbers; if evidence is thin say so.`; let report = ""; let usage = { input: 0, output: 0, cached: 0, reasoning: 0 }; const t0 = Date.now(); for await (const ev of adapterFor(model.provider).stream({ model, messages: [{ role: "user", parts: [{ type: "text", text: synthPrompt }] }], settings: { maxOutputTokens: 16_000, effort: depth === "deep" ? "high" : "medium", reasoningSummary: false }, signal: o.signal, loadAttachment: async () => null })) { if (ev.type === "text") { report += ev.delta; o.emit("text", { delta: ev.delta }); } else if (ev.type === "usage") usage = { input: usage.input + ev.usage.input, output: usage.output + ev.usage.output, cached: usage.cached + (ev.usage.cached ?? 0), reasoning: usage.reasoning + (ev.usage.reasoning ?? 0) }; else if (ev.type === "error") throw new Error(ev.message); } const synthCost = estimateCost(model, usage) ?? 0; cost += synthCost; await recordUsage({ provider: model.provider, modelId: model.id, kind: "research", conversationId: conv.id, projectId: o.projectId ?? null, messageId: assistant.id, usage, costEstimated: synthCost, latencyMs: Date.now() - t0, meta: { researchId: id } }); const srcOut = sources.map(({ text: _t, ...s }) => s); await sql`UPDATE research_runs SET status = 'complete', report_md = ${report}, sources = ${sql.json(srcOut as never)}, cost_usd = ${cost}, completed_at = now(), steps = ${sql.json(steps as never)} WHERE id = ${id}`; await updateMessage(assistant.id, { parts: [{ type: "text", text: report }, ...srcOut.filter(s => s.fetched).map(s => ({ type: "citation" as const, url: s.url, title: s.title, snippet: s.snippet.slice(0, 200) }))], status: "complete", usage: usage as unknown as Record, costUsd: cost, latencyMs: Date.now() - t0 }); o.emit("done", { id, conversationId: conv.id, costUsd: cost, sources: srcOut.length }); } catch (e) { const msg = (e as Error).message; await sql`UPDATE research_runs SET status = ${o.signal?.aborted ? "cancelled" : "error"}, error = ${msg}, cost_usd = ${cost} WHERE id = ${id}`; await updateMessage(assistant.id, { status: "error", error: msg }); o.emit("error", { message: msg }); } return { id, conversationId: conv.id }; }