import type { Id } from "./_generated/dataModel"; import type { MutationCtx } from "./_generated/server"; interface ProjectionContext { readonly assistantMessageId: Id<"conversationMessages"> | null; readonly turnId: Id<"conversationTurns"> | null; } const parseStringField = (record: unknown, key: string): string | undefined => { if (typeof record !== "object" || record === null) { return undefined; } const value = (record as Record)[key]; return typeof value === "string" ? value : undefined; }; const resolveContext = async ( ctx: MutationCtx, submissionId: string | undefined ): Promise => { if (submissionId === undefined) { return { assistantMessageId: null, turnId: null }; } const turn = await ctx.db .query("conversationTurns") .withIndex("by_submissionId", (q) => q.eq("submissionId", submissionId)) .unique(); if (turn === null) { return { assistantMessageId: null, turnId: null }; } const assistant = await ctx.db .query("conversationMessages") .withIndex("by_turnId_and_role", (q) => q.eq("turnId", turn._id).eq("role", "assistant") ) .unique(); return { assistantMessageId: assistant?._id ?? null, turnId: turn._id, }; }; const terminalError = (record: unknown): string => { if (typeof record !== "object" || record === null) { return "Flue submission failed"; } const value = (record as Record).error; if (typeof value === "string") { return value.slice(0, 2000); } if (typeof value === "object" && value !== null) { const { message } = value as Record; if (typeof message === "string") { return message.slice(0, 2000); } } return "Flue submission failed"; }; /** * Project the small product view from Flue's canonical conversation stream. * The raw stream remains authoritative; this function only updates the existing * assistant message and turn row that the web already reads. */ export const projectConversationRecords = async ( ctx: MutationCtx, recordsJson: string, submissionId: string | undefined ): Promise => { let records: unknown[]; try { const parsed = JSON.parse(recordsJson) as unknown; if (!Array.isArray(parsed)) { return; } records = parsed; } catch { return; } const projection = await resolveContext(ctx, submissionId); if (projection.turnId === null) { return; } for (const record of records) { const type = parseStringField(record, "type"); if (type === "assistant_text_delta" && projection.assistantMessageId) { const delta = parseStringField(record, "delta"); if (delta !== undefined) { const assistant = await ctx.db.get(projection.assistantMessageId); if (assistant !== null) { await ctx.db.patch(assistant._id, { content: `${assistant.content}${delta}`, }); } } continue; } if (type !== "submission_settled") { continue; } const turn = await ctx.db.get(projection.turnId); if ( turn === null || turn.status === "completed" || turn.status === "failed" || turn.status === "aborted" ) { continue; } const outcome = parseStringField(record, "outcome"); if (outcome === "completed" && projection.assistantMessageId) { const result = typeof record === "object" && record !== null ? (record as Record).result : undefined; const text = typeof result === "object" && result !== null ? (result as Record).text : undefined; if (typeof text === "string") { await ctx.db.patch(projection.assistantMessageId, { content: text }); } } let status: "aborted" | "completed" | "failed" = "failed"; if (outcome === "completed") { status = "completed"; } else if (outcome === "aborted") { status = "aborted"; } await ctx.db.patch(turn._id, { completedAt: Date.now(), error: outcome === "failed" ? terminalError(record) : undefined, leaseExpiresAt: undefined, leaseOwner: undefined, status, }); } };