MEGAINFRA-WEBAPP/src/app/api/chat/route.ts

/**
 * POST /api/chat — streams an answer as Server-Sent Events.
 *
 * Route Handlers use the Web Request/Response API, so streaming is a
 * plain ReadableStream in a Response. POST is never cached.
 *
 * GATED: returns 404 unless ENABLE_CHAT=true. There is no authentication
 * on this route beyond `x-ask-user` identification — anyone with the PIN
 * can pick any name until logins land (see src/lib/agent/context.ts).
 *
 * Two request shapes:
 *   - New: `{ threadId?, message, display?, attachmentIds?, images? }`
 *     with an `x-ask-user` header. No threadId creates a thread owned by
 *     the actor; a thread owned by someone else 404s. The conversation is
 *     persisted to `ask_messages` so it survives leaving the page.
 *   - Legacy: `{ history }` with no `threadId` — kept working for one
 *     release so a tab open across a deploy doesn't break. Not persisted.
 */
import { tools, type AgentTool } from "@/lib/agent/tools";
import { actionTools } from "@/lib/agent/tools-actions";
import { listMemories, formatMemoriesForPrompt } from "@/lib/agent/memory";
import { extractCharts } from "@/lib/reports/chart-spec";
import type { ChartSpec } from "@/lib/reports/chart-spec";
import { resolveActor } from "@/lib/agent/context";
import type { AgentContext } from "@/lib/agent/context";
import {
  autoTitle,
  createThread,
  getThreadOwner,
  loadThreadEntries,
  appendMessage,
  getRecentActionIds,
} from "@/lib/chat/threads";
import { extractActionCard, describeActionOutcome, type ActionCard } from "@/lib/agent/gateway";
import { extractGeneratedImages } from "@/lib/agent/images";
import { extractPdfMeta, type PdfReportMeta } from "@/lib/agent/pdf-report";
import {
  buildSystemPrompt,
  chatEnabled,
  getProvider,
  runChatLoop,
  toChatToolDefs,
  type ChatEntry,
  type ChatImage,
  type ChatToolCall,
  type ChatToolResult,
} from "@/lib/chat";

export const dynamic = "force-dynamic";

/**
 * Coerce whatever the browser posted into well-formed ChatEntry values.
 *
 * This is a trust boundary: the request body is JSON, so the ChatEntry
 * type guarantees nothing at runtime. The client legitimately sends
 * assistant turns as {role, text} with no toolCalls, which previously
 * crashed both adapters on the second message of a conversation.
 */
function normalizeHistory(raw: unknown): ChatEntry[] {
  if (!Array.isArray(raw)) return [];
  const out: ChatEntry[] = [];

  for (const item of raw) {
    if (!item || typeof item !== "object") continue;
    const e = item as Record<string, unknown>;

    if (e.role === "user") {
      out.push({
        role: "user",
        text: String(e.text ?? ""),
        ...(Array.isArray(e.images) ? { images: e.images as ChatImage[] } : {}),
      });
    } else if (e.role === "assistant") {
      out.push({
        role: "assistant",
        text: String(e.text ?? ""),
        toolCalls: Array.isArray(e.toolCalls)
          ? (e.toolCalls as ChatToolCall[])
          : [],
      });
    } else if (e.role === "tool") {
      out.push({
        role: "tool",
        results: Array.isArray(e.results) ? (e.results as ChatToolResult[]) : [],
      });
    }
  }

  return out;
}

interface ChatRequestBody {
  history?: unknown;
  threadId?: string;
  message?: string;
  display?: string;
  attachmentIds?: string[];
  images?: ChatImage[];
}

/**
 * Runs the agent loop and streams the result. When `threadId`/`ownerId`
 * are given, the finished (or failed) assistant turn is persisted as its
 * own `ask_messages` row — see the plan's append-only rule.
 */
function streamAnswer(args: {
  history: ChatEntry[];
  threadId?: string;
  ownerId?: string;
  threadTitle?: string;
  system?: string;
  extraTools?: AgentTool[];
}) {
  const { history, threadId, ownerId, threadTitle } = args;
  const provider = getProvider();
  const allTools = [...tools, ...(args.extraTools ?? [])];
  const toolByName = new Map(allTools.map((t) => [t.name, t]));
  const toolDefs = toChatToolDefs(allTools);
  const system = args.system ?? buildSystemPrompt(new Date());
  const encoder = new TextEncoder();

  const stream = new ReadableStream({
    async start(controller) {
      const send = (event: string, data: unknown) => {
        controller.enqueue(
          encoder.encode(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`),
        );
      };

      const collectedCharts: ChartSpec[] = [];
      const collectedActionIds: string[] = [];
      const toolsUsed: string[] = [];
      let lastText = "";
      let pdfMeta: PdfReportMeta | null = null;

      try {
        send("meta", { provider: provider.name, model: provider.model });
        if (threadId && threadTitle) send("thread", { id: threadId, title: threadTitle });

        const result = await runChatLoop({
          provider,
          history,
          tools: toolDefs,
          system,
          onTextDelta: (delta) => {
            lastText += delta;
            send("delta", { text: delta });
          },
          executeTool: async (name, input) => {
            send("tool", { name });
            toolsUsed.push(name);
            const tool = toolByName.get(name);
            if (!tool) throw new Error(`Unknown tool: ${name}`);
            const raw = JSON.stringify(await tool.handler(input));
            // An action card (propose_changes) goes to the browser in
            // full; the model keeps only actionId/status/problems — the
            // same split extractCharts makes for chart numbers.
            const { forModel: withoutCard, card } = extractActionCard(raw);
            if (card) {
              collectedActionIds.push(card.actionId);
              send("action", card as ActionCard);
            }
            // Generated images go to the browser in full; the model
            // keeps only their ids, so it can place [[image:<id>]]
            // without ever seeing pixels or a URL.
            const { forModel: withoutImages, images } = extractGeneratedImages(withoutCard);
            for (const img of images) send("image", img);

            // Chart specs go to the browser; the model gets { id, title }
            // only, so it can place a chart but never re-derive its numbers.
            const { forModel: withoutCharts, charts } = extractCharts(withoutImages);
            for (const c of charts) {
              collectedCharts.push(c);
              send("chart", c);
            }

            // make_pdf_report — the model's own "this answer is a
            // finished report" signal, called last. The browser gets
            // the full title/subtitle/filename; the model just an ack.
            const { forModel, pdf } = extractPdfMeta(withoutCharts);
            if (pdf) {
              pdfMeta = pdf;
              send("pdf", pdf);
            }
            return forModel;
          },
        });

        let messageId: string | undefined;
        if (threadId && ownerId) {
          messageId = await appendMessage(threadId, ownerId, {
            kind: "assistant",
            text: result.text,
            charts: collectedCharts,
            tools: toolsUsed,
            actionIds: collectedActionIds,
            ...(pdfMeta ? { parsed: { pdf: pdfMeta } } : {}),
          });
        }

        send("done", {
          text: result.text,
          iterations: result.iterations,
          stoppedOnLimit: result.stoppedOnLimit,
          ...(messageId ? { messageId } : {}),
        });
      } catch (err) {
        // Surface the reason. A silent close would leave the UI showing a
        // half-finished answer with no indication it failed.
        const message = err instanceof Error ? err.message : String(err);
        if (threadId && ownerId) {
          try {
            // Persist what happened so far — a crash must not silently
            // drop the partial answer. "error" in tools marks the turn
            // as having failed mid-stream.
            await appendMessage(threadId, ownerId, {
              kind: "assistant",
              text: lastText,
              charts: collectedCharts,
              tools: [...toolsUsed, "error"],
            });
          } catch {
            // Best-effort — the error event below still reaches the client.
          }
        }
        send("error", { message });
      } finally {
        controller.close();
      }
    },
  });

  return new Response(stream, {
    headers: {
      "Content-Type": "text/event-stream; charset=utf-8",
      "Cache-Control": "no-cache, no-transform",
      Connection: "keep-alive",
    },
  });
}

export async function POST(request: Request) {
  if (!chatEnabled()) {
    return new Response("Not found", { status: 404 });
  }

  let body: ChatRequestBody;
  try {
    body = (await request.json()) as ChatRequestBody;
  } catch {
    return Response.json({ error: "Invalid JSON body" }, { status: 400 });
  }

  // Legacy path: an open tab from before threads shipped sends { history }
  // with no threadId. Keep it working for one release; it is never
  // persisted, matching its pre-threads behaviour exactly.
  if (body.history !== undefined && body.threadId === undefined) {
    const history = normalizeHistory(body.history);
    if (history.length === 0) {
      return Response.json({ error: "history is empty" }, { status: 400 });
    }
    return streamAnswer({ history });
  }

  const actor = await resolveActor(request);
  if (actor instanceof Response) return actor;

  const message = typeof body.message === "string" ? body.message : "";
  if (!message.trim()) {
    return Response.json({ error: "message is empty" }, { status: 400 });
  }

  let threadId = body.threadId;
  let threadTitle: string | undefined;

  if (threadId) {
    const owner = await getThreadOwner(threadId);
    if (owner === null || owner !== actor.id) {
      return Response.json({ error: "Thread not found" }, { status: 404 });
    }
  } else {
    const created = await createThread(actor.id, autoTitle(message));
    threadId = created.id;
    threadTitle = created.title;
  }

  const priorEntries = await loadThreadEntries(threadId);

  // "The next user turn gets a system note": ChatEntry has no system
  // role, so an outcome for the last turn's action(s) — confirmed,
  // rejected, undone or failed — is prepended to what the model reads
  // as this user turn, once, since the next assistant reply will carry
  // its own (possibly empty) action_ids and this lookup moves forward.
  let systemNote = "";
  try {
    const recentActionIds = await getRecentActionIds(threadId);
    const notes = (
      await Promise.all(recentActionIds.map((id) => describeActionOutcome(id)))
    ).filter((n): n is string => !!n);
    if (notes.length > 0) systemNote = `${notes.join("\n")}\n\n`;
  } catch {
    // ask_log may not exist yet — proceed without the note.
  }

  const images = Array.isArray(body.images) ? body.images : undefined;
  const userEntry: ChatEntry = {
    role: "user",
    text: systemNote + message,
    ...(images && images.length > 0 ? { images } : {}),
  };
  const history = [...priorEntries, userEntry];

  // Created before the model is ever called — a crash mid-turn must never
  // lose what was actually asked.
  await appendMessage(threadId, actor.id, {
    kind: "user",
    text: message,
    ...(body.display ? { display: body.display } : {}),
    ...(body.attachmentIds ? { attachmentIds: body.attachmentIds } : {}),
  });

  const ctx: AgentContext = { actor, threadId };
  // Memories and action tools both need the actor — the legacy path above
  // has none, so it gets neither, matching its pre-threads behaviour.
  let memoriesBlock = "";
  try {
    memoriesBlock = formatMemoriesForPrompt(await listMemories(ctx));
  } catch {
    // ask_log may not exist yet (schema not applied) — answer without
    // memories rather than failing the whole turn.
  }

  return streamAnswer({
    history,
    threadId,
    ownerId: actor.id,
    threadTitle,
    system: buildSystemPrompt(new Date(), memoriesBlock),
    extraTools: actionTools(ctx),
  });
}