import { requireUser } from "@/lib/auth"; import { analystConfig, jevKey, jevModel } from "@/lib/settings"; import { digestPlan, digestForModel, PlanError } from "@/lib/digest"; import { proposeCandidates, streamReport, AnalystError } from "@/lib/analyst"; import { judgeCandidates, jevFallback, makeJevClient } from "@/lib/jev"; import { sse } from "@/lib/stream"; import { newAnalysis, patchAnalysis } from "@/lib/db"; import { reportPrompt } from "@/lib/report-spec"; export const runtime = "nodejs"; export const dynamic = "force-dynamic"; const MAX_BODY_BYTES = 16_000_000; export async function POST(req: Request) { const user = await requireUser().catch(() => null); if (!user) return Response.json({ error: "Sign in first." }, { status: 401 }); const analyst = analystConfig(user.id); const jev = jevKey(user.id); if (!analyst) return Response.json({ error: "Set the analyst model in Settings first." }, { status: 412 }); if (!jev) return Response.json({ error: "Set your Jev (TypeSafe) API key in Settings first." }, { status: 412 }); const raw = await req.text(); if (raw.length > MAX_BODY_BYTES) return Response.json({ error: "Request too large." }, { status: 413 }); let xml = ""; let note = ""; try { const body = JSON.parse(raw); xml = String(body.xml ?? ""); note = String(body.note ?? "").slice(0, 500); } catch { return Response.json({ error: "Expected JSON with an xml field." }, { status: 400 }); } const id = newAnalysis(user.id, note.trim().slice(0, 120) || "Plan analysis", xml); const enc = new TextEncoder(); const prompt = reportPrompt(); // The analyst model needs minutes per stage. If the DBA closes the tab, enqueue() does not throw, so the // producer would keep firing model calls into a dead stream and queue behind its own ghost. cancel() is the // only signal that the reader is gone: stop launching further calls and abort the one in flight. const ac = new AbortController(); let abandoned = false; const stream = new ReadableStream({ async start(ctrl) { const send = (event: string, data: unknown) => ctrl.enqueue(enc.encode(sse(event, data))); let oul = ""; const close = (status: string, error?: string) => { patchAnalysis(id, { status, ...(error ? { error } : {}), ...(oul ? { oul } : {}) }); send("done", { id }); ctrl.close(); }; try { send("stage", { stage: "digesting" }); let digest; try { digest = digestPlan(xml); } catch (e) { const m = e instanceof PlanError ? e.message : "Could not read that plan."; send("error", { stage: "digesting", message: m }); patchAnalysis(id, { status: "failed", error: m }); send("done", { id }); return ctrl.close(); } if (abandoned) return giveUp(id, oul); patchAnalysis(id, { digest: JSON.stringify(digest) }); send("stage", { stage: "proposing" }); let candidates; try { candidates = await proposeCandidates(analyst, digest, note, fetch, ac.signal); } catch (e) { const m = e instanceof AnalystError ? e.message : "The analyst model call failed."; send("error", { stage: "proposing", message: m }); patchAnalysis(id, { status: "failed", error: m }); send("done", { id }); return ctrl.close(); } if (abandoned) return giveUp(id, oul); candidates.forEach((c) => send("candidate", { key: c.key, title: c.title })); patchAnalysis(id, { candidates: JSON.stringify(candidates) }); send("stage", { stage: "judging" }); let verdict; try { verdict = await judgeCandidates(digest, candidates, makeJevClient(jev, jevModel(user.id))); } catch (e) { verdict = jevFallback(candidates, `Jev call failed: ${(e as Error).message}`); } send("verdict", verdict); patchAnalysis(id, { verdict: JSON.stringify(verdict) }); send("stage", { stage: "writing" }); for await (const chunk of streamReport(analyst, { plan: digestForModel(digest), options: candidates, verdict }, prompt, fetch, ac.signal)) { oul += chunk; send("oul", { chunk }); } if (abandoned) return giveUp(id, oul); close(oul ? "done" : "partial", oul ? undefined : "The analyst model returned an empty report."); } catch (e) { const m = abandoned ? "Browser stopped listening." : ((e as Error)?.message ?? "Analysis failed."); if (!abandoned) send("error", { stage: "stream", message: m }); patchAnalysis(id, { status: "failed", error: m, ...(oul ? { oul } : {}) }); send("done", { id }); ctrl.close(); } function giveUp(id: string, oul: string) { patchAnalysis(id, { status: oul ? "partial" : "abandoned", error: "Closed before the run finished.", ...(oul ? { oul } : {}) }); try { ctrl.close(); } catch { /* already closed */ } } }, cancel() { abandoned = true; ac.abort(); } }); return new Response(stream, { headers: { "content-type": "text/event-stream; charset=utf-8", "cache-control": "no-store", connection: "keep-alive", "x-accel-buffering": "no" } }); }