feat: analysis pipeline with staged SSE

Adds @openuidev/lang-core (one-line reason: the prompt-side library must build in a non-React Node runtime; react-lang's createContext crashes the route's server bundle).

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

app/api/analyze/route.ts

@@ -0,0 +1,95 @@
+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<Uint8Array>({ 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" } });
+}