import { randomUUID } from "node:crypto"; import type { IncomingMessage, ServerResponse } from "node:http"; import { abortReroll, beginReroll, createConversation, createShell, deleteConversation, deleteShell, duplicateConversation, EMPTY_SHELL_SLOTS, getConversation, getShell, listConversations, listShells, mergeTurnReplies, promoteTemporary, runDirectChatMultiTurn, saveConversation, switchTurnVariant, updateConversation, updateShell, writeRerollReplies, type DirectChatAssistantReply, type DirectChatConversation, type DirectChatMessage, type DirectChatMultiTurnResult, type DirectChatMultiWindowResult, type DirectChatShellSlots, type DirectChatStreamDelta, type DirectChatTurn, type DirectChatWindow, } from "../direct-chat/index.js"; import { listProfilesForCompare } from "./model-compare.js"; async function readBody(req: IncomingMessage): Promise { const chunks: Buffer[] = []; for await (const chunk of req) { chunks.push(chunk as Buffer); } return Buffer.concat(chunks).toString("utf8"); } function json(res: ServerResponse, status: number, data: unknown): void { res.writeHead(status, { "Content-Type": "application/json; charset=utf-8" }); res.end(JSON.stringify(data)); } function readWindows(raw: unknown): DirectChatWindow[] | undefined { if (!Array.isArray(raw)) return undefined; const out: DirectChatWindow[] = []; for (const item of raw) { if (!item || typeof item !== "object" || Array.isArray(item)) continue; const row = item as Record; const id = typeof row.id === "string" ? row.id.trim() : ""; const profileId = typeof row.profileId === "string" ? row.profileId.trim() : ""; if (!id || !profileId) continue; const window: DirectChatWindow = { id, profileId }; if (typeof row.profileName === "string" && row.profileName.trim()) { window.profileName = row.profileName.trim(); } if (typeof row.model === "string" && row.model.trim()) { window.model = row.model.trim(); } out.push(window); } return out; } function wantsNdjson(req: IncomingMessage): boolean { return String(req.headers.accept || "").includes("application/x-ndjson"); } type MultiStreamHooks = { onDelta: (event: DirectChatStreamDelta) => void; onWindow: (result: DirectChatMultiWindowResult) => void; }; function replyFromWindowResult( result: DirectChatMultiWindowResult, ): DirectChatAssistantReply { return { windowId: result.windowId, content: result.reply ?? "", ...(result.reasoning ? { thinking: result.reasoning } : {}), ...(result.model ? { model: result.model } : {}), ...(result.error ? { error: result.error } : {}), ...(result.contextTrace ? { contextTrace: result.contextTrace } : {}), }; } /** 流式增量按窗口累加,并定期写入会话,避免刷新后丢掉已生成的正文 */ function createStreamingSaver( conversation: DirectChatConversation, windows: DirectChatWindow[], text: string, turnId: string, baseTurns?: DirectChatTurn[], ) { const ts = new Date().toISOString(); const replies: DirectChatAssistantReply[] = []; let timer: ReturnType | null = null; let closed = false; let produced = false; let turnsState = baseTurns ? structuredClone(baseTurns) : null; const saveNow = () => { if (timer) { clearTimeout(timer); timer = null; } if (closed) return; if (turnsState) { turnsState = writeRerollReplies( turnsState, turnId, replies.map((r) => ({ ...r })), ); saveConversation({ ...conversation, windows, turns: turnsState, }); return; } const turn: DirectChatTurn = { id: turnId, user: { content: text, createdAt: ts }, replies: replies.map((r) => ({ ...r })), createdAt: ts, }; saveConversation({ ...conversation, windows, turns: [...conversation.turns.filter((t) => t.id !== turnId), turn], }); }; const saveSoon = () => { if (closed || timer) return; timer = setTimeout(() => { timer = null; saveNow(); }, 400); }; return { onDelta(event: DirectChatStreamDelta) { let reply = replies.find((r) => r.windowId === event.windowId); if (!reply) { reply = { windowId: event.windowId, content: "" }; replies.push(reply); } if (event.delta) produced = true; if (event.field === "thinking") { reply.thinking = `${reply.thinking || ""}${event.delta}`; } else { reply.content += event.delta; } saveSoon(); }, onWindow(result: DirectChatMultiWindowResult) { const next = replyFromWindowResult(result); if (next.content || next.thinking || next.error) produced = true; const idx = replies.findIndex((r) => r.windowId === result.windowId); if (idx >= 0) replies[idx] = next; else replies.push(next); saveNow(); }, sawOutput() { return produced; }, currentTurns() { return turnsState; }, close() { closed = true; if (timer) { clearTimeout(timer); timer = null; } }, }; } async function writeMultiStream( res: ServerResponse, run: (hooks: MultiStreamHooks) => Promise, finish: (multi: DirectChatMultiTurnResult) => Record, ): Promise< | { ok: true; multi: DirectChatMultiTurnResult } | { ok: false; error: string } > { res.writeHead(200, { "Content-Type": "application/x-ndjson; charset=utf-8", "Cache-Control": "no-cache, no-transform", "X-Accel-Buffering": "no", }); const writeEvent = (event: unknown) => { res.write(`${JSON.stringify(event)}\n`); }; try { const multi = await run({ onDelta: (event) => writeEvent({ type: "delta", ...event }), onWindow: (result) => writeEvent({ type: "window", result }), }); writeEvent({ type: "done", multi, ...finish(multi) }); res.end(); return { ok: true, multi }; } catch (err) { const error = err instanceof Error ? err.message : String(err); try { writeEvent({ type: "error", error }); res.end(); } catch { res.destroy(); } return { ok: false, error }; } } function slotsFromBody(body: Record): DirectChatShellSlots { return { systemPrompt: typeof body.systemPrompt === "string" ? body.systemPrompt : EMPTY_SHELL_SLOTS.systemPrompt, loreBefore: typeof body.loreBefore === "string" ? body.loreBefore : EMPTY_SHELL_SLOTS.loreBefore, seedHistory: typeof body.seedHistory === "string" ? body.seedHistory : EMPTY_SHELL_SLOTS.seedHistory, seedMessages: Array.isArray(body.seedMessages) ? (body.seedMessages as DirectChatShellSlots["seedMessages"]) : EMPTY_SHELL_SLOTS.seedMessages, loreAfter: typeof body.loreAfter === "string" ? body.loreAfter : EMPTY_SHELL_SLOTS.loreAfter, postTurn: typeof body.postTurn === "string" ? body.postTurn : EMPTY_SHELL_SLOTS.postTurn, charName: typeof body.charName === "string" ? body.charName : undefined, exampleDialogue: typeof body.exampleDialogue === "string" ? body.exampleDialogue : undefined, charDescription: typeof body.charDescription === "string" ? body.charDescription : undefined, charPersonality: typeof body.charPersonality === "string" ? body.charPersonality : undefined, scenario: typeof body.scenario === "string" ? body.scenario : undefined, }; } export async function handleDirectChatApi( req: IncomingMessage, res: ServerResponse, pathname: string, ): Promise { if (!pathname.startsWith("/api/direct-chat")) return false; try { if (req.method === "GET" && pathname === "/api/direct-chat/shells") { json(res, 200, { shells: listShells() }); return true; } if (req.method === "POST" && pathname === "/api/direct-chat/shells") { const body = JSON.parse(await readBody(req)) as Record; const shell = createShell({ name: typeof body.name === "string" ? body.name : undefined, slots: slotsFromBody(body), presetId: body.presetId === null ? null : typeof body.presetId === "string" ? body.presetId : undefined, }); json(res, 201, { shell }); return true; } const shellMatch = pathname.match( /^\/api\/direct-chat\/shells\/([^/]+)(\/.*)?$/, ); if (shellMatch) { const shellId = decodeURIComponent(shellMatch[1]!); const sub = shellMatch[2] ?? ""; if (req.method === "GET" && sub === "") { const shell = getShell(shellId); if (!shell) { json(res, 404, { error: "上下文壳不存在" }); return true; } json(res, 200, { shell, conversations: listConversations(shellId), }); return true; } if ( (req.method === "PATCH" || req.method === "PUT") && sub === "" ) { const body = JSON.parse(await readBody(req)) as Record; try { const shell = updateShell(shellId, { name: typeof body.name === "string" ? body.name : undefined, systemPrompt: typeof body.systemPrompt === "string" ? body.systemPrompt : undefined, loreBefore: typeof body.loreBefore === "string" ? body.loreBefore : undefined, seedHistory: typeof body.seedHistory === "string" ? body.seedHistory : undefined, seedMessages: Array.isArray(body.seedMessages) ? (body.seedMessages as DirectChatShellSlots["seedMessages"]) : undefined, loreAfter: typeof body.loreAfter === "string" ? body.loreAfter : undefined, postTurn: typeof body.postTurn === "string" ? body.postTurn : undefined, presetId: body.presetId === null ? null : typeof body.presetId === "string" ? body.presetId : undefined, }); json(res, 200, { shell }); } catch (err) { json(res, 404, { error: err instanceof Error ? err.message : String(err), }); } return true; } if (req.method === "DELETE" && sub === "") { try { deleteShell(shellId); json(res, 200, { ok: true }); } catch (err) { json(res, 404, { error: err instanceof Error ? err.message : String(err), }); } return true; } if (req.method === "GET" && sub === "/conversations") { if (!getShell(shellId)) { json(res, 404, { error: "上下文壳不存在" }); return true; } json(res, 200, { conversations: listConversations(shellId) }); return true; } if (req.method === "POST" && sub === "/conversations") { const body = JSON.parse(await readBody(req)) as { name?: string; messages?: DirectChatMessage[]; windows?: DirectChatWindow[]; turns?: DirectChatTurn[]; }; try { const conversation = createConversation({ shellId, name: body.name, messages: body.messages, windows: readWindows(body.windows), turns: Array.isArray(body.turns) ? body.turns : undefined, }); json(res, 201, { conversation }); } catch (err) { json(res, 400, { error: err instanceof Error ? err.message : String(err), }); } return true; } } const convMatch = pathname.match( /^\/api\/direct-chat\/conversations\/([^/]+)(\/.*)?$/, ); if (convMatch) { const conversationId = decodeURIComponent(convMatch[1]!); const sub = convMatch[2] ?? ""; if (req.method === "GET" && sub === "") { const conversation = getConversation(conversationId); if (!conversation) { json(res, 404, { error: "会话不存在" }); return true; } const shell = getShell(conversation.shellId); json(res, 200, { conversation, shell }); return true; } if ( (req.method === "PATCH" || req.method === "PUT") && sub === "" ) { const body = JSON.parse(await readBody(req)) as { name?: string; messages?: DirectChatMessage[]; windows?: DirectChatWindow[]; }; try { const conversation = updateConversation(conversationId, { name: body.name, messages: body.messages, windows: readWindows(body.windows), }); json(res, 200, { conversation }); } catch (err) { json(res, 404, { error: err instanceof Error ? err.message : String(err), }); } return true; } if (req.method === "DELETE" && sub === "") { try { deleteConversation(conversationId); json(res, 200, { ok: true }); } catch (err) { json(res, 404, { error: err instanceof Error ? err.message : String(err), }); } return true; } if (req.method === "POST" && sub === "/duplicate") { try { const conversation = duplicateConversation(conversationId); json(res, 201, { conversation }); } catch (err) { json(res, 404, { error: err instanceof Error ? err.message : String(err), }); } return true; } if (req.method === "POST" && sub === "/messages") { const body = JSON.parse(await readBody(req)) as { text?: string; windows?: DirectChatWindow[]; complete?: boolean; }; const text = typeof body.text === "string" ? body.text.trim() : ""; if (!text) { json(res, 400, { error: "text 不能为空" }); return true; } const conversation = getConversation(conversationId); if (!conversation) { json(res, 404, { error: "会话不存在" }); return true; } const shell = getShell(conversation.shellId); if (!shell) { json(res, 404, { error: "上下文壳不存在" }); return true; } const windows = Array.isArray(body.windows) && body.windows.length ? body.windows : conversation.windows; if (!windows.length) { json(res, 400, { error: "请先新增至少一个模型窗口" }); return true; } const turnId = randomUUID(); const saver = createStreamingSaver(conversation, windows, text, turnId); const run = (hooks?: MultiStreamHooks) => runDirectChatMultiTurn({ message: text, slots: { systemPrompt: shell.systemPrompt, loreBefore: shell.loreBefore, seedHistory: shell.seedHistory, seedMessages: shell.seedMessages, loreAfter: shell.loreAfter, postTurn: shell.postTurn, }, presetId: shell.presetId, windows, turns: conversation.turns, complete: body.complete, turnId, onDelta: (event) => { saver.onDelta(event); hooks?.onDelta(event); }, onWindow: (result) => { saver.onWindow(result); hooks?.onWindow(result); }, }); try { if (wantsNdjson(req)) { await writeMultiStream(res, run, (multi) => { saver.close(); const updated = saveConversation({ ...conversation, windows, turns: [ ...conversation.turns.filter((t) => t.id !== multi.turn.id), multi.turn, ], }); return { conversation: updated, shell }; }); return true; } const multi = await run(); saver.close(); const updated = saveConversation({ ...conversation, windows, turns: [ ...conversation.turns.filter((t) => t.id !== multi.turn.id), multi.turn, ], }); json(res, 200, { conversation: updated, shell, multi, }); } catch (err) { json(res, 400, { error: err instanceof Error ? err.message : String(err), }); } return true; } const rerollMatch = sub.match(/^\/turns\/([^/]+)\/reroll$/); if (rerollMatch && req.method === "POST") { const turnId = decodeURIComponent(rerollMatch[1]!); const body = JSON.parse(await readBody(req)) as { text?: string; windows?: DirectChatWindow[]; complete?: boolean; }; const text = typeof body.text === "string" ? body.text.trim() : ""; if (!text) { json(res, 400, { error: "text 不能为空" }); return true; } const conversation = getConversation(conversationId); if (!conversation) { json(res, 404, { error: "会话不存在" }); return true; } const shell = getShell(conversation.shellId); if (!shell) { json(res, 404, { error: "上下文壳不存在" }); return true; } const windows = Array.isArray(body.windows) && body.windows.length ? body.windows : conversation.windows; if (!windows.length) { json(res, 400, { error: "请先新增至少一个模型窗口" }); return true; } const prepared = beginReroll(conversation.turns, turnId, text); if (!prepared) { json(res, 404, { error: "回合不存在" }); return true; } const preparedConversation = saveConversation({ ...conversation, windows, turns: prepared.turns, }); const saver = createStreamingSaver( preparedConversation, windows, text, turnId, prepared.turns, ); const run = (hooks?: MultiStreamHooks) => runDirectChatMultiTurn({ message: text, slots: { systemPrompt: shell.systemPrompt, loreBefore: shell.loreBefore, seedHistory: shell.seedHistory, seedMessages: shell.seedMessages, loreAfter: shell.loreAfter, postTurn: shell.postTurn, }, presetId: shell.presetId, windows, turns: prepared.prior, complete: body.complete, turnId, onDelta: (event) => { saver.onDelta(event); hooks?.onDelta(event); }, onWindow: (result) => { saver.onWindow(result); hooks?.onWindow(result); }, }); const restoreIfEmpty = () => { saver.close(); if (saver.sawOutput()) return; const restored = abortReroll(prepared.turns, turnId); if (!restored) return; saveConversation({ ...preparedConversation, windows, turns: restored, }); }; try { if (wantsNdjson(req)) { const outcome = await writeMultiStream(res, run, (multi) => { saver.close(); const turns = writeRerollReplies( saver.currentTurns() ?? prepared.turns, turnId, multi.turn.replies, ); const updated = saveConversation({ ...preparedConversation, windows, turns, }); return { conversation: updated, shell }; }); if (!outcome.ok) restoreIfEmpty(); return true; } const multi = await run(); saver.close(); const turns = writeRerollReplies( saver.currentTurns() ?? prepared.turns, turnId, multi.turn.replies, ); const updated = saveConversation({ ...preparedConversation, windows, turns, }); json(res, 200, { conversation: updated, shell, multi }); } catch (err) { restoreIfEmpty(); json(res, 400, { error: err instanceof Error ? err.message : String(err), }); } return true; } const retryMatch = sub.match(/^\/turns\/([^/]+)\/retry$/); if (retryMatch && req.method === "POST") { const turnId = decodeURIComponent(retryMatch[1]!); const body = JSON.parse(await readBody(req)) as { windowIds?: string[]; complete?: boolean; }; const requested = Array.isArray(body.windowIds) ? body.windowIds.filter( (x): x is string => typeof x === "string" && Boolean(x.trim()), ) : []; if (!requested.length) { json(res, 400, { error: "缺少 windowIds" }); return true; } const conversation = getConversation(conversationId); if (!conversation) { json(res, 404, { error: "会话不存在" }); return true; } const shell = getShell(conversation.shellId); if (!shell) { json(res, 404, { error: "上下文壳不存在" }); return true; } const turnIndex = conversation.turns.findIndex((t) => t.id === turnId); const turn = turnIndex >= 0 ? conversation.turns[turnIndex] : undefined; if (!turn) { json(res, 404, { error: "回合不存在" }); return true; } const text = turn.user.content.trim(); if (!text) { json(res, 400, { error: "本轮缺少用户输入" }); return true; } const wanted = new Set(requested); const windows = conversation.windows.filter((w) => wanted.has(w.id)); if (!windows.length) { json(res, 400, { error: "没有可重试的窗口" }); return true; } try { // 关键:不调 beginReroll。重试只补选中的窗口,其余回复原地保留。 const runRetry = (hooks?: Partial) => runDirectChatMultiTurn({ message: text, slots: { systemPrompt: shell.systemPrompt, loreBefore: shell.loreBefore, seedHistory: shell.seedHistory, seedMessages: shell.seedMessages, loreAfter: shell.loreAfter, postTurn: shell.postTurn, }, presetId: shell.presetId, windows, turns: conversation.turns.slice(0, turnIndex), complete: body.complete, onDelta: hooks?.onDelta, onWindow: hooks?.onWindow, }); // 每完成一个窗口就落盘:多个窗口是并行的,先好的先写入, // 前端也能逐张刷新,不用等最慢的那个。 const persistOne = (result: DirectChatMultiWindowResult) => { const latest = getConversation(conversationId); if (!latest) return; const reply: DirectChatAssistantReply = { windowId: result.windowId, content: result.reply ?? "", ...(result.reasoning ? { thinking: result.reasoning } : {}), ...(result.model ? { model: result.model } : {}), ...(result.error ? { error: result.error } : {}), ...(result.contextTrace ? { contextTrace: result.contextTrace } : {}), }; saveConversation({ ...latest, turns: mergeTurnReplies(latest.turns, turnId, [reply]), }); }; if (wantsNdjson(req)) { const outcome = await writeMultiStream( res, (hooks) => runRetry({ onDelta: hooks.onDelta, onWindow: (result) => { persistOne(result); hooks.onWindow(result); }, }), () => ({ conversation: getConversation(conversationId) ?? conversation, shell, }), ); if (!outcome.ok) return true; return true; } const multi = await runRetry({ onWindow: (result) => persistOne(result), }); const latest = getConversation(conversationId) ?? conversation; const updated = saveConversation({ ...latest, turns: mergeTurnReplies(latest.turns, turnId, multi.turn.replies), }); json(res, 200, { conversation: updated, shell, multi }); } catch (err) { json(res, 400, { error: err instanceof Error ? err.message : String(err), }); } return true; } const variantMatch = sub.match(/^\/turns\/([^/]+)\/variant$/); if (variantMatch && req.method === "POST") { const turnId = decodeURIComponent(variantMatch[1]!); const body = JSON.parse(await readBody(req)) as { direction?: string }; if (body.direction !== "prev" && body.direction !== "next") { json(res, 400, { error: "direction 须为 prev 或 next" }); return true; } const conversation = getConversation(conversationId); if (!conversation) { json(res, 404, { error: "会话不存在" }); return true; } const turns = switchTurnVariant( conversation.turns, turnId, body.direction, ); if (!turns) { json(res, 400, { error: "没有更多版本" }); return true; } const updated = saveConversation({ ...conversation, turns }); json(res, 200, { conversation: updated }); return true; } const selectMatch = sub.match(/^\/turns\/([^/]+)\/select$/); if (selectMatch && req.method === "POST") { const turnId = decodeURIComponent(selectMatch[1]!); const body = JSON.parse(await readBody(req)) as { windowId?: string | null; }; const conversation = getConversation(conversationId); if (!conversation) { json(res, 404, { error: "会话不存在" }); return true; } const windowId = body.windowId === null || body.windowId === undefined ? null : String(body.windowId).trim() || null; if (windowId) { const turn = conversation.turns.find((t) => t.id === turnId); if (!turn?.replies.some((r) => r.windowId === windowId)) { json(res, 400, { error: "该窗口没有本轮回复" }); return true; } } const turns = conversation.turns.map((t) => t.id === turnId ? { ...t, selectedWindowId: windowId } : t, ); const updated = saveConversation({ ...conversation, turns }); json(res, 200, { conversation: updated }); return true; } } if ( req.method === "POST" && pathname === "/api/direct-chat/ephemeral/complete" ) { const body = JSON.parse(await readBody(req)) as Record & { text?: string; complete?: boolean; presetId?: string | null; windows?: DirectChatWindow[]; turns?: DirectChatTurn[]; }; const text = typeof body.text === "string" ? body.text.trim() : ""; if (!text) { json(res, 400, { error: "text 不能为空" }); return true; } const windows = Array.isArray(body.windows) ? body.windows : []; const turns = Array.isArray(body.turns) ? body.turns : []; if (!windows.length) { json(res, 400, { error: "请先新增至少一个模型窗口" }); return true; } const run = (hooks?: MultiStreamHooks) => runDirectChatMultiTurn({ message: text, slots: slotsFromBody(body), presetId: body.presetId === null ? null : typeof body.presetId === "string" ? body.presetId : undefined, windows, turns, complete: body.complete, onDelta: hooks?.onDelta, onWindow: hooks?.onWindow, }); try { if (wantsNdjson(req)) { await writeMultiStream(res, run, () => ({})); return true; } const multi = await run(); json(res, 200, { multi }); } catch (err) { json(res, 400, { error: err instanceof Error ? err.message : String(err), }); } return true; } if (req.method === "POST" && pathname === "/api/direct-chat/promote") { const body = JSON.parse(await readBody(req)) as { shellId?: string; shell?: { name?: string; systemPrompt?: string; loreBefore?: string; seedHistory?: string; seedMessages?: DirectChatShellSlots["seedMessages"]; loreAfter?: string; postTurn?: string; presetId?: string | null; }; conversationName?: string; messages?: DirectChatMessage[]; windows?: DirectChatWindow[]; turns?: DirectChatTurn[]; }; try { const promoted = promoteTemporary({ shellId: body.shellId, shell: body.shell ? { name: body.shell.name, slots: { systemPrompt: body.shell.systemPrompt ?? "", loreBefore: body.shell.loreBefore ?? "", seedHistory: body.shell.seedHistory ?? "", seedMessages: body.shell.seedMessages ?? [], loreAfter: body.shell.loreAfter ?? "", postTurn: body.shell.postTurn ?? "", }, presetId: body.shell.presetId, } : undefined, conversationName: body.conversationName, messages: Array.isArray(body.messages) ? body.messages : [], windows: Array.isArray(body.windows) ? body.windows : [], turns: Array.isArray(body.turns) ? body.turns : [], }); json(res, 201, promoted); } catch (err) { json(res, 400, { error: err instanceof Error ? err.message : String(err), }); } return true; } if ( req.method === "GET" && pathname === "/api/direct-chat/compare-profiles" ) { json(res, 200, { profiles: listProfilesForCompare() }); return true; } return false; } catch (err) { json(res, 500, { error: err instanceof Error ? err.message : String(err), }); return true; } }