Allow message retries for "pending" status, refactor message context handling, remove workspace import/export routes, and add SQLite-backed rate limiting.
CI / check (22) (push) Failing after 16m26s
CI / check (24) (push) Failing after 15m49s
CI / check (22) (pull_request) Failing after 16m26s
CI / check (24) (pull_request) Failing after 15m50s

This commit is contained in:
2026-10-09 19:28:43 +02:00
parent 8666e09049
commit f67d918d09
13 changed files with 313 additions and 193 deletions
+6 -2
View File
@@ -12,10 +12,12 @@ import {createConfig} from "./config.js";
import {createDatabase} from "./db/index.js";
import {createRepositories} from "./db/repositories.js";
import {createAuthRateLimiters, createChatRateLimiters} from "./middleware/rateLimiters.js";
import {createSqliteRateLimitStore} from "./middleware/rateLimit.js";
import {createModelService} from "./services/modelService.js";
import {createChatService} from "./services/chatService.js";
import {createAuthRouter} from "./routes/auth.js";
import {createChatsRouter} from "./routes/chats.js";
import {createWorkspaceTransferRouter} from "./routes/workspaceTransfer.js";
import {createPagesRouter} from "./routes/pages.js";
import {createSettingsRouter} from "./routes/settings.js";
import {createHealthRouter, createSystemRouter} from "./routes/system.js";
@@ -272,8 +274,9 @@ export function createApp(options = {}) {
const attachmentService = createAttachmentService(db, config, modelService, visionDebugService);
const chatService = createChatService(repositories, modelService, config, undefined, attachmentService, visionDebugService);
const FileStore = FileStoreFactory(session);
const authRateLimiters = createAuthRateLimiters(config);
const chatRateLimiters = createChatRateLimiters(config);
const rateLimitStore = createSqliteRateLimitStore(db);
const authRateLimiters = createAuthRateLimiters(config, rateLimitStore);
const chatRateLimiters = createChatRateLimiters(config, rateLimitStore);
closeHandlers.push(
...authRateLimiters.map((limiter) => limiter.close),
...chatRateLimiters.map((limiter) => limiter.close),
@@ -334,6 +337,7 @@ export function createApp(options = {}) {
if (config.visionDebug) app.use(createVisionDebugRouter({visionDebugService, repositories}));
app.use(createSettingsRouter({repositories}));
app.use(createChatsRouter({repositories, chatService, modelService, config, chatRateLimiters, attachmentService}));
app.use(createWorkspaceTransferRouter({repositories, chatService, config, attachmentService}));
app.use(createSystemRouter({modelService, config}));
app.use(notFoundHandler);
app.use(errorHandler);
+27 -2
View File
@@ -8,7 +8,7 @@ function pruneBucket(bucket, windowMs, now) {
}
}
export function createRateLimiter({windowMs, max, keyGenerator, message, cleanupIntervalMs = 60 * 1000}) {
export function createRateLimiter({windowMs, max, keyGenerator, message, cleanupIntervalMs = 60 * 1000, store = null}) {
const hits = new Map();
let cleanupTimer = null;
@@ -22,7 +22,7 @@ export function createRateLimiter({windowMs, max, keyGenerator, message, cleanup
}
};
if (cleanupIntervalMs > 0) {
if (!store && cleanupIntervalMs > 0) {
cleanupTimer = setInterval(cleanup, cleanupIntervalMs);
// Der Aufräumtimer allein soll den Node-Prozess beim Herunterfahren nicht offen halten.
cleanupTimer.unref();
@@ -35,6 +35,10 @@ export function createRateLimiter({windowMs, max, keyGenerator, message, cleanup
}
const now = Date.now();
if (store) {
if (!store.consume(key, now, windowMs, max)) return res.status(429).json({error: message});
return next();
}
const bucket = hits.get(key) || [];
pruneBucket(bucket, windowMs, now);
if (!bucket.length) {
@@ -59,3 +63,24 @@ export function createRateLimiter({windowMs, max, keyGenerator, message, cleanup
return limiter;
}
// SQLite is already shared by all application instances and serializes this
// short transaction, so the check-and-increment remains atomic across workers.
export function createSqliteRateLimitStore(db) {
db.exec(`CREATE TABLE IF NOT EXISTS rate_limit_hits (
bucket_key TEXT NOT NULL,
hit_at INTEGER NOT NULL
);
CREATE INDEX IF NOT EXISTS rate_limit_hits_expiry ON rate_limit_hits(hit_at);`);
db.pragma("busy_timeout = 5000");
const prune = db.prepare("DELETE FROM rate_limit_hits WHERE bucket_key = ? AND hit_at <= ?");
const count = db.prepare("SELECT COUNT(*) AS count FROM rate_limit_hits WHERE bucket_key = ?");
const insert = db.prepare("INSERT INTO rate_limit_hits (bucket_key, hit_at) VALUES (?, ?)");
const consume = db.transaction((key, now, windowMs, max) => {
prune.run(key, now - windowMs);
if (count.get(key).count >= max) return false;
insert.run(key, now);
return true;
});
return {consume};
}
+6 -2
View File
@@ -3,12 +3,13 @@
*/
import {createRateLimiter} from "./rateLimit.js";
export function createAuthRateLimiters(config) {
export function createAuthRateLimiters(config, store = null) {
return [
createRateLimiter({
windowMs: config.authRateLimitWindowMs,
max: config.authRateLimitMaxPerIp,
cleanupIntervalMs: config.rateLimitCleanupIntervalMs,
store,
keyGenerator: (req) => `auth:ip:${req.ip}`,
message: "Too many authentication attempts from this IP. Please try again later."
}),
@@ -16,6 +17,7 @@ export function createAuthRateLimiters(config) {
windowMs: config.authRateLimitWindowMs,
max: config.authRateLimitMaxPerAccount,
cleanupIntervalMs: config.rateLimitCleanupIntervalMs,
store,
keyGenerator: (req) => {
const username = String(req.body?.username || "").trim().toLowerCase();
return username ? `auth:user:${username}` : null;
@@ -25,12 +27,13 @@ export function createAuthRateLimiters(config) {
];
}
export function createChatRateLimiters(config) {
export function createChatRateLimiters(config, store = null) {
return [
createRateLimiter({
windowMs: config.chatRateLimitWindowMs,
max: config.chatRateLimitMaxPerIp,
cleanupIntervalMs: config.rateLimitCleanupIntervalMs,
store,
keyGenerator: (req) => `chat:ip:${req.ip}`,
message: "Too many chat requests from this IP. Please slow down."
}),
@@ -38,6 +41,7 @@ export function createChatRateLimiters(config) {
windowMs: config.chatRateLimitWindowMs,
max: config.chatRateLimitMaxPerAccount,
cleanupIntervalMs: config.rateLimitCleanupIntervalMs,
store,
keyGenerator: (req) => req.session.user ? `chat:user:${req.session.user}` : null,
message: "Too many chat requests for this account. Please slow down."
})
+11 -127
View File
@@ -5,11 +5,8 @@ import express from "express";
import {createTextStreamEvents} from "../services/streamEvents.js";
import {asyncHandler} from "../middleware/asyncHandler.js";
import {generationCancellation} from "../middleware/generationCancellation.js";
import multer from "multer";
import {requireLogin} from "../middleware/auth.js";
import {validateBody} from "../middleware/validate.js";
import {validateWorkspaceImport} from "../services/workspaceImportService.js";
import {createWorkspaceArchive, parseWorkspaceArchive, MAX_ARCHIVE_BYTES} from "../services/workspaceArchive.js";
import {getPersonality, listPersonalities, DEFAULT_PERSONALITY_ID} from "../services/personalityRegistry.js";
const createChatValidator = (body) => ({
@@ -36,11 +33,13 @@ const sendMessageValidator = (body) => {
if (!Array.isArray(attachments) || attachments.some(id => typeof id !== "string" || !/^[a-f0-9-]{36}$/.test(id))) return {error: "Invalid attachment IDs"};
const message = String(body?.message || "").trim() || (attachments.length ? "Describe the attached images." : "");
const chatId = Number(body?.chatId);
const userMessageId = body?.userMessageId == null ? null : Number(body.userMessageId);
if (userMessageId != null && (!Number.isInteger(userMessageId) || userMessageId < 1)) return {error: "Invalid message ID"};
if (!chatId) return {error: "chatId is required"};
if (!message) return {error: "message is required"};
const preferences = searchPreferencesValidator(body);
if (preferences.error) return preferences;
return {value: {message, chatId, attachments, model: String(body?.model || "qwen3:14b"), ...preferences.value}};
return {value: {message, chatId, attachments, userMessageId, model: String(body?.model || "qwen3:14b"), ...preferences.value}};
};
const organizationValidator = (body) => ({
@@ -124,29 +123,9 @@ const userPersonaValidator = (body) => {
return {value: {userPersonaId}};
};
export function createChatsRouter({repositories, chatService, modelService, config, chatRateLimiters, attachmentService}) {
export function createChatsRouter({repositories, chatService, modelService, config, chatRateLimiters}) {
const router = express.Router();
router.use(requireLogin);
const receiveArchive = multer({storage: multer.memoryStorage(), limits: {fileSize: MAX_ARCHIVE_BYTES, files: 1, parts: 2}}).single("archive");
const readArchive = (req, res, next) => receiveArchive(req, res, error => {
if (error) return res.status(error.code === "LIMIT_FILE_SIZE" ? 413 : 400).json({error: "Invalid or oversized workspace archive"});
try {
if (!req.file) throw new Error("Choose a workspace archive");
req.archive = parseWorkspaceArchive(req.file.buffer);
next();
} catch (failure) { res.status(400).json({error: failure.message}); }
});
const archiveImages = (workspace, files) => {
const images = new Map();
for (const chat of workspace.chats) for (const message of chat.messages) for (const attachment of message.attachments) {
if (images.has(attachment.id)) continue;
const matches = [...files.entries()].filter(([name]) => name.startsWith(`images/${attachment.id}.`));
if (matches.length !== 1) throw new Error(`Image ${attachment.id} is missing from archive`);
const [name, buffer] = matches[0];
images.set(attachment.id, {buffer, size: buffer.length, originalname: name.split("/").at(-1)});
}
return images;
};
router.use(["/chats/:id/messages/by-index/:index", "/chats/:id/messages/:messageId"], (req, res, next) => {
const chatId = Number(req.params.id);
const messageId = req.params.messageId ? Number(req.params.messageId) : chatService.getChatMessageByIndex(chatId, req.params.index)?.id;
@@ -496,8 +475,12 @@ export function createChatsRouter({repositories, chatService, modelService, conf
if (!chat) return res.status(404).json({error: "Chat not found"});
const payload = chatService.buildMessagesPayload({user: req.session.user, chatId, session: chat});
const characters = payload.messagesPayload.reduce((total, message) => total + String(message.content || "").length, 0);
const summaryMessageId = Number(chat.context_summary_message_id || 0);
const summarySources = summaryMessageId ? chatService.listActiveBranchMessages(chatId)
.filter(message => message.id <= summaryMessageId && message.deliveryStatus === "complete")
.map(({id, role, content}) => ({id, role, content})) : [];
return res.json({memories: repositories.listChatMemories(chatId, req.session.user), summary: chat.context_summary || "",
summaryMessageId: chat.context_summary_message_id || 0, summaryMaxChars: config.chatSummaryMaxChars,
summaryMessageId, summarySources, summaryMaxChars: config.chatSummaryMaxChars,
context: {estimatedTokens: Math.ceil(characters / 4), contextLength: chat.context_length || null,
recentMessageLimit: config.chatHistoryLimit, memoryFactLimit: 30}});
});
@@ -566,6 +549,7 @@ export function createChatsRouter({repositories, chatService, modelService, conf
chatId: req.validatedBody.chatId,
message: req.validatedBody.message,
attachments: req.validatedBody.attachments,
userMessageId: req.validatedBody.userMessageId,
model: req.validatedBody.model,
webSearch: req.validatedBody.webSearch,
reasoning: req.validatedBody.reasoning
@@ -599,6 +583,7 @@ export function createChatsRouter({repositories, chatService, modelService, conf
chatId: req.validatedBody.chatId,
message: req.validatedBody.message,
attachments: req.validatedBody.attachments,
userMessageId: req.validatedBody.userMessageId,
model: req.validatedBody.model,
...createTextStreamEvents(sendEvent),
webSearch: req.validatedBody.webSearch,
@@ -708,107 +693,6 @@ export function createChatsRouter({repositories, chatService, modelService, conf
}
}));
router.get("/exports/chats/:id", (req, res) => {
const chatId = Number(req.params.id);
const format = String(req.query.format || "txt").toLowerCase();
const chat = repositories.getChat(chatId, req.session.user);
if (!chat) return res.status(404).json({error: "Chat not found"});
const messages = repositories.listChatMessages(chatId).map(chatService.formatChatMessage);
if (format === "json") {
return res.json({chat, messages});
}
if (format === "md") {
const body = [`# ${chat.title}`, "", ...messages.flatMap((message) => [`## ${message.role === "user" ? "You" : "Assistant"}`, "", message.content, ""])].join("\n");
res.type("text/markdown");
return res.send(body);
}
const body = [`Chat: ${chat.title}`, "", ...messages.flatMap((message) => [`${message.role === "user" ? "You" : "Assistant"}:`, message.content, ""])].join("\n");
res.type("text/plain");
return res.send(body);
});
router.get("/exports/workspace", (req, res) => res.json({workspace: repositories.exportWorkspace(req.session.user)}));
router.get("/exports/workspace/archive", asyncHandler(async (req, res) => {
try {
const workspace = repositories.exportWorkspace(req.session.user);
const images = new Map();
for (const chat of workspace.chats) for (const message of chat.messages) for (const attachment of message.attachments) {
if (images.has(attachment.id)) continue;
const row = attachmentService.owned(attachment.id, req.session.user);
images.set(attachment.id, [`images/${row.id}.${row.mime_type.split("/")[1]}`, await attachmentService.storage.read(row.storage_key)]);
}
const archive = createWorkspaceArchive(workspace, [...images.values()]);
res.set({"Content-Type": "application/gzip", "Content-Disposition": 'attachment; filename="workspace-backup.tar.gz"', "Cache-Control": "private, no-store"});
return res.send(archive);
} catch (error) {
return res.status(error.message.includes("250 MB") ? 413 : 500).json({error: "Workspace archive could not be created"});
}
}));
router.post("/imports/workspace/archive/preview", readArchive, (req, res) => {
const validation = validateWorkspaceImport(req.archive.workspace, repositories.exportWorkspace(req.session.user));
if (validation.ok) {
try { archiveImages(validation.workspace, req.archive.images); }
catch (error) { validation.ok = false; validation.errors.push(error.message); }
}
return res.status(validation.ok ? 200 : 400).json({ok: validation.ok, errors: validation.errors, warnings: validation.warnings, preview: validation.preview});
});
router.post("/imports/workspace/archive", readArchive, asyncHandler(async (req, res) => {
const validation = validateWorkspaceImport(req.archive.workspace, repositories.exportWorkspace(req.session.user));
if (!validation.ok) return res.status(400).json({error: "Workspace import validation failed", errors: validation.errors});
if (validation.preview.duplicates.length && req.body?.allowDuplicates !== "true") {
return res.status(409).json({error: "Workspace import contains duplicate records", warnings: validation.warnings, preview: validation.preview});
}
let images;
try { images = archiveImages(validation.workspace, req.archive.images); }
catch (error) { return res.status(400).json({error: error.message}); }
const restored = [];
try {
const ids = new Map();
for (const [oldId, file] of images) {
const uploaded = await attachmentService.upload(file, req.session.user, {forImport: true});
restored.push(uploaded.id);
ids.set(oldId, uploaded.id);
}
for (const chat of validation.workspace.chats) for (const message of chat.messages) {
message.attachments = message.attachments.map(attachment => ({id: ids.get(attachment.id)}));
}
const imported = repositories.importWorkspace(req.session.user, validation.workspace);
return res.json({imported, preview: validation.preview});
} catch (error) {
for (const id of restored) await attachmentService.remove(id, req.session.user).catch(() => {});
return res.status(error.status || 400).json({error: error.message || "Workspace archive import failed"});
}
}));
router.post("/imports/workspace/preview", (req, res) => {
const validation = validateWorkspaceImport(req.body?.workspace, repositories.exportWorkspace(req.session.user));
return res.status(validation.ok ? 200 : 400).json({
ok: validation.ok,
errors: validation.errors,
warnings: validation.warnings,
preview: validation.preview
});
});
router.post("/imports/workspace", (req, res) => {
const validation = validateWorkspaceImport(req.body?.workspace, repositories.exportWorkspace(req.session.user));
if (!validation.ok) {
return res.status(400).json({
error: "Workspace import validation failed",
errors: validation.errors,
warnings: validation.warnings,
preview: validation.preview
});
}
if (validation.preview.duplicates.length && !req.body?.allowDuplicates) {
return res.status(409).json({
error: "Workspace import contains duplicate records",
errors: [],
warnings: validation.warnings,
preview: validation.preview
});
}
return res.json({imported: repositories.importWorkspace(req.session.user, validation.workspace), preview: validation.preview});
});
router.get("/chat/history", (req, res) => res.json({history: repositories.listChats(req.session.user)}));
return router;
+92
View File
@@ -0,0 +1,92 @@
import express from "express";
import multer from "multer";
import {requireLogin} from "../middleware/auth.js";
import {asyncHandler} from "../middleware/asyncHandler.js";
import {validateWorkspaceImport} from "../services/workspaceImportService.js";
import {createWorkspaceArchive, parseWorkspaceArchive, MAX_ARCHIVE_BYTES} from "../services/workspaceArchive.js";
export function createWorkspaceTransferRouter({repositories, chatService, attachmentService}) {
const router = express.Router();
router.use(requireLogin);
const receiveArchive = multer({storage: multer.memoryStorage(), limits: {fileSize: MAX_ARCHIVE_BYTES, files: 1, parts: 2}}).single("archive");
const readArchive = (req, res, next) => receiveArchive(req, res, error => {
if (error) return res.status(error.code === "LIMIT_FILE_SIZE" ? 413 : 400).json({error: "Invalid or oversized workspace archive"});
try {
if (!req.file) throw new Error("Choose a workspace archive");
req.archive = parseWorkspaceArchive(req.file.buffer);
next();
} catch (failure) { res.status(400).json({error: failure.message}); }
});
const archiveImages = (workspace, files) => {
const images = new Map();
for (const chat of workspace.chats) for (const message of chat.messages) for (const attachment of message.attachments) {
if (images.has(attachment.id)) continue;
const matches = [...files.entries()].filter(([name]) => name.startsWith(`images/${attachment.id}.`));
if (matches.length !== 1) throw new Error(`Image ${attachment.id} is missing from archive`);
const [name, buffer] = matches[0];
images.set(attachment.id, {buffer, size: buffer.length, originalname: name.split("/").at(-1)});
}
return images;
};
router.get("/exports/chats/:id", (req, res) => {
const chatId = Number(req.params.id), format = String(req.query.format || "txt").toLowerCase();
const chat = repositories.getChat(chatId, req.session.user);
if (!chat) return res.status(404).json({error: "Chat not found"});
const messages = repositories.listChatMessages(chatId).map(chatService.formatChatMessage);
if (format === "json") return res.json({chat, messages});
const markdown = format === "md";
const body = markdown
? [`# ${chat.title}`, "", ...messages.flatMap(message => [`## ${message.role === "user" ? "You" : "Assistant"}`, "", message.content, ""])].join("\n")
: [`Chat: ${chat.title}`, "", ...messages.flatMap(message => [`${message.role === "user" ? "You" : "Assistant"}:`, message.content, ""])].join("\n");
res.type(markdown ? "text/markdown" : "text/plain");
return res.send(body);
});
router.get("/exports/workspace", (req, res) => res.json({workspace: repositories.exportWorkspace(req.session.user)}));
router.get("/exports/workspace/archive", asyncHandler(async (req, res) => {
try {
const workspace = repositories.exportWorkspace(req.session.user), images = new Map();
for (const chat of workspace.chats) for (const message of chat.messages) for (const attachment of message.attachments) {
if (images.has(attachment.id)) continue;
const row = attachmentService.owned(attachment.id, req.session.user);
images.set(attachment.id, [`images/${row.id}.${row.mime_type.split("/")[1]}`, await attachmentService.storage.read(row.storage_key)]);
}
const archive = createWorkspaceArchive(workspace, [...images.values()]);
res.set({"Content-Type": "application/gzip", "Content-Disposition": 'attachment; filename="workspace-backup.tar.gz"', "Cache-Control": "private, no-store"});
return res.send(archive);
} catch (error) { return res.status(error.message.includes("250 MB") ? 413 : 500).json({error: "Workspace archive could not be created"}); }
}));
router.post("/imports/workspace/archive/preview", readArchive, (req, res) => {
const validation = validateWorkspaceImport(req.archive.workspace, repositories.exportWorkspace(req.session.user));
if (validation.ok) try { archiveImages(validation.workspace, req.archive.images); } catch (error) { validation.ok = false; validation.errors.push(error.message); }
return res.status(validation.ok ? 200 : 400).json({ok: validation.ok, errors: validation.errors, warnings: validation.warnings, preview: validation.preview});
});
router.post("/imports/workspace/archive", readArchive, asyncHandler(async (req, res) => {
const validation = validateWorkspaceImport(req.archive.workspace, repositories.exportWorkspace(req.session.user));
if (!validation.ok) return res.status(400).json({error: "Workspace import validation failed", errors: validation.errors});
if (validation.preview.duplicates.length && req.body?.allowDuplicates !== "true") return res.status(409).json({error: "Workspace import contains duplicate records", warnings: validation.warnings, preview: validation.preview});
let images;
try { images = archiveImages(validation.workspace, req.archive.images); } catch (error) { return res.status(400).json({error: error.message}); }
const restored = [];
try {
const ids = new Map();
for (const [oldId, file] of images) { const uploaded = await attachmentService.upload(file, req.session.user, {forImport: true}); restored.push(uploaded.id); ids.set(oldId, uploaded.id); }
for (const chat of validation.workspace.chats) for (const message of chat.messages) message.attachments = message.attachments.map(attachment => ({id: ids.get(attachment.id)}));
return res.json({imported: repositories.importWorkspace(req.session.user, validation.workspace), preview: validation.preview});
} catch (error) {
for (const id of restored) await attachmentService.remove(id, req.session.user).catch(() => {});
return res.status(error.status || 400).json({error: error.message || "Workspace archive import failed"});
}
}));
router.post("/imports/workspace/preview", (req, res) => {
const validation = validateWorkspaceImport(req.body?.workspace, repositories.exportWorkspace(req.session.user));
return res.status(validation.ok ? 200 : 400).json({ok: validation.ok, errors: validation.errors, warnings: validation.warnings, preview: validation.preview});
});
router.post("/imports/workspace", (req, res) => {
const validation = validateWorkspaceImport(req.body?.workspace, repositories.exportWorkspace(req.session.user));
if (!validation.ok) return res.status(400).json({error: "Workspace import validation failed", errors: validation.errors, warnings: validation.warnings, preview: validation.preview});
if (validation.preview.duplicates.length && !req.body?.allowDuplicates) return res.status(409).json({error: "Workspace import contains duplicate records", errors: [], warnings: validation.warnings, preview: validation.preview});
return res.json({imported: repositories.importWorkspace(req.session.user, validation.workspace), preview: validation.preview});
});
return router;
}
+30 -53
View File
@@ -10,9 +10,6 @@ import {
containsUserVoiceInRoleplayOpener
} from "./personaService.js";
import {
buildContextSummaryPrompt,
buildMemoryPrompt,
buildPersonaStatePrompt,
clampRetryActiveIndex,
formatChatMessage as formatBaseChatMessage,
formatMessagesForSummary,
@@ -21,10 +18,10 @@ import {
} from "./chatMessageHelpers.js";
import {createHash} from "node:crypto";
import { SearXNGSearchProvider, WebSearchError, buildWebSearchContext, decideWebSearch, retrieveWebSearch } from "./webSearchService.js";
import { SearXNGSearchProvider, WebSearchError, decideWebSearch, retrieveWebSearch } from "./webSearchService.js";
import { supportsReasoning } from "../../public/js/modelCapabilities.js";
import {VISION_EVIDENCE_GUIDANCE} from "./visionAnalysis.js";
import {getPersonalitySystemPrompt} from "./personalityRegistry.js";
import {createConversationContextBuilder} from "./conversationContext.js";
export function createChatService(repositories, modelService, config, webSearchProvider = new SearXNGSearchProvider(config), attachmentService = null, visionDebugService = null) {
const formatChatMessage = (row, related = null) => {
@@ -298,45 +295,7 @@ export function createChatService(repositories, modelService, config, webSearchP
return repositories.getActiveUserPersona(user);
};
const buildConversationContext = ({ user, chatId, session = null, message, upToMessageId = null, sources = null, historyOverride = null, visionSelection = null, contextSummaryOverride = undefined }) => {
const activeSession = session || repositories.getChat(chatId, user);
if (!activeSession) return null;
const activePersona = repositories.getAssistantPersonaForChat(chatId, user);
const activeUserPersona = getUserPersonaForSession(chatId, user, activeSession, activePersona);
const history = historyOverride || (upToMessageId == null
? repositories.getRecentChatMessages(chatId).reverse()
: repositories.getRecentChatMessagesUpToId(chatId, upToMessageId).reverse());
const conversation = message == null ? history : [...history, { role: "user", content: message }];
const messagesPayload = [];
// Result material remains a separate untrusted user-context message. The
// grounding rule is added after mutable persona/chat prompts so it remains
// the final trusted instruction governing the supplied references.
const webSearchContext = sources === null ? null : buildWebSearchContext(sources, now());
if (activeSession.system_instruction) {
messagesPayload.push({ role: "system", content: `CHAT-SPECIFIC INSTRUCTIONS:\n${activeSession.system_instruction}` });
}
messagesPayload.push({role: "system", content: getPersonalitySystemPrompt(activeSession.personality_id)});
const contextSummaryPrompt = buildContextSummaryPrompt(contextSummaryOverride === undefined
? (upToMessageId == null ? activeSession.context_summary : null) : contextSummaryOverride);
if (contextSummaryPrompt) messagesPayload.push({ role: "system", content: contextSummaryPrompt });
const memoryPrompt = buildMemoryPrompt(repositories.listChatMemories(chatId, user));
if (memoryPrompt) messagesPayload.push({ role: "system", content: memoryPrompt });
const personaStatePrompt = buildPersonaStatePrompt(repositories.getPersonaChatState(chatId, user));
if (personaStatePrompt) messagesPayload.push({ role: "system", content: `Conversation continuity notes (context only):\n${personaStatePrompt}` });
const conversationMessages = conversation.map((entry) => ({
role: entry.role === "bot" ? "assistant" : "user",
content: entry.content
}));
messagesPayload.push(...conversationMessages.slice(0, -1));
const currentQuestion = conversation.at(-1)?.content || "";
const visionContext = attachmentService?.context(visionSelection || attachmentService.selectImages(
message == null ? history : [...history, {role: "user", content: message}], currentQuestion), currentQuestion);
if (visionContext) messagesPayload.push(...visionContext);
if (webSearchContext) messagesPayload.push(webSearchContext[0]);
if (webSearchContext?.[1]) messagesPayload.push(webSearchContext[1]);
if (conversationMessages.length) messagesPayload.push(conversationMessages.at(-1));
return { session: activeSession, activePersona, activeUserPersona, messagesPayload };
};
const buildConversationContext = createConversationContextBuilder({repositories, attachmentService, now});
const buildRetryPayload = async ({ chatId, user, session, selectedModel, targetMessage, retryStyle = null, signal = null, generation = {}, webSearch = true, reasoning = false, onChunk = null, onThinking = null, onStatus = null, requestId = null }) => {
let thinking = "";
@@ -432,7 +391,7 @@ export function createChatService(repositories, modelService, config, webSearchP
return { fullReply, thinking, sources, searchState, promptMessageId: promptMessage.id };
};
const generateChatTurn = ({ user, chatId, message, model, attachments = [], reuseChatId = null, webSearch = true, reasoning = false, onStatus = null, onChunk = null, onThinking = null, signal = null, parentMessageId = undefined, requestId = null }) =>
const generateChatTurn = ({ user, chatId, message, model, attachments = [], reuseChatId = null, userMessageId: resumeMessageId = null, webSearch = true, reasoning = false, onStatus = null, onChunk = null, onThinking = null, signal = null, parentMessageId = undefined, requestId = null }) =>
withChatGenerationLock(chatId, async () => {
await (summaryQueues.get(chatId) || Promise.resolve());
const session = repositories.getChat(chatId, user);
@@ -442,7 +401,18 @@ export function createChatService(repositories, modelService, config, webSearchP
session.context_summary = null;
session.context_summary_message_id = 0;
}
const effectiveParentId = parentMessageId === undefined
const resumedMessage = resumeMessageId == null ? null : repositories.getChatMessage(chatId, resumeMessageId);
if (resumeMessageId != null && (!resumedMessage || resumedMessage.role !== "user" || !["pending", "failed"].includes(resumedMessage.delivery_status))) {
return {error: "This message cannot be resumed", status: 409};
}
if (resumedMessage?.delivery_status === "pending") {
const createdAt = Date.parse(`${String(resumedMessage.created_at).replace(" ", "T")}Z`);
const staleAfterMs = Number(config.modelMaxGenerationMs || 600000) + 60000;
if (Number.isFinite(createdAt) && new Date(now()).getTime() - createdAt < staleAfterMs) {
return {error: "This generation may still be running. Wait a little longer before resuming it.", status: 409};
}
}
const effectiveParentId = resumedMessage ? resumedMessage.parent_message_id : parentMessageId === undefined
? session.active_leaf_message_id ?? repositories.getLatestChatMessage(chatId)?.id ?? null
: parentMessageId;
if (effectiveParentId != null && !repositories.getChatMessage(chatId, effectiveParentId)) {
@@ -451,9 +421,16 @@ export function createChatService(repositories, modelService, config, webSearchP
const { model: effectiveModel, generation } = getGenerationSettings(session, model);
const useReasoning = reasoning === true && supportsReasoning(effectiveModel);
signal?.throwIfAborted();
const insert = () => repositories.insertBranchChatMessage(chatId, "user", message, null, null, "pending", null, effectiveParentId);
const inserted = attachmentService ? attachmentService.createMessage(attachments, user, chatId, insert, reuseChatId) : insert();
const userMessageId = Number(inserted?.lastInsertRowid);
let userMessageId;
if (resumedMessage) {
userMessageId = Number(resumedMessage.id);
repositories.updateChatMessageDelivery(chatId, userMessageId, "pending", null);
repositories.updateChatActiveLeaf(chatId, user, userMessageId);
} else {
const insert = () => repositories.insertBranchChatMessage(chatId, "user", message, null, null, "pending", null, effectiveParentId);
const inserted = attachmentService ? attachmentService.createMessage(attachments, user, chatId, insert, reuseChatId) : insert();
userMessageId = Number(inserted?.lastInsertRowid);
}
try {
let sources = null;
let searchState = "skipped";
@@ -677,11 +654,11 @@ export function createChatService(repositories, modelService, config, webSearchP
},
persistChatMessageRetryState,
buildMessagesPayload: buildConversationContext,
async sendChatMessage({ user, chatId, message, model, attachments = [], webSearch = true, reasoning = false, signal = null, requestId = null }) {
return generateChatTurn({ user, chatId, message, model, attachments, webSearch, reasoning, signal, requestId });
async sendChatMessage({ user, chatId, message, model, attachments = [], userMessageId = null, webSearch = true, reasoning = false, signal = null, requestId = null }) {
return generateChatTurn({ user, chatId, message, model, attachments, userMessageId, webSearch, reasoning, signal, requestId });
},
async streamChatMessage({ user, chatId, message, model, attachments = [], webSearch = true, reasoning = false, onStatus, onChunk, onThinking, signal = null, requestId = null }) {
return generateChatTurn({ user, chatId, message, model, attachments, webSearch, reasoning, onStatus, onChunk, onThinking, signal, requestId });
async streamChatMessage({ user, chatId, message, model, attachments = [], userMessageId = null, webSearch = true, reasoning = false, onStatus, onChunk, onThinking, signal = null, requestId = null }) {
return generateChatTurn({ user, chatId, message, model, attachments, userMessageId, webSearch, reasoning, onStatus, onChunk, onThinking, signal, requestId });
},
isGenerating: (chatId) => chatGenerationQueues.has(chatId),
isRegeneratingMessage: (chatId, messageId) => regeneratingMessages.get(chatId) === messageId,
+68
View File
@@ -0,0 +1,68 @@
import {
buildContextSummaryPrompt,
buildMemoryPrompt,
buildPersonaStatePrompt
} from "./chatMessageHelpers.js";
import {buildWebSearchContext} from "./webSearchService.js";
import {getPersonalitySystemPrompt} from "./personalityRegistry.js";
export function createConversationContextBuilder({repositories, attachmentService, now}) {
const getUserPersonaForSession = (chatId, user, activePersona) => {
const chatPersona = repositories.getUserPersonaForChat(chatId, user);
if (chatPersona || activePersona) return chatPersona || null;
return repositories.getActiveUserPersona(user);
};
return ({
user,
chatId,
session = null,
message,
upToMessageId = null,
sources = null,
historyOverride = null,
visionSelection = null,
contextSummaryOverride = undefined
}) => {
const activeSession = session || repositories.getChat(chatId, user);
if (!activeSession) return null;
const activePersona = repositories.getAssistantPersonaForChat(chatId, user);
const activeUserPersona = getUserPersonaForSession(chatId, user, activePersona);
const history = historyOverride || (upToMessageId == null
? repositories.getRecentChatMessages(chatId).reverse()
: repositories.getRecentChatMessagesUpToId(chatId, upToMessageId).reverse());
const conversation = message == null ? history : [...history, {role: "user", content: message}];
const messagesPayload = [];
// Search results stay untrusted user context, followed by their grounding rule.
const webSearchContext = sources === null ? null : buildWebSearchContext(sources, now());
if (activeSession.system_instruction) messagesPayload.push({
role: "system",
content: `CHAT-SPECIFIC INSTRUCTIONS:\n${activeSession.system_instruction}`
});
messagesPayload.push({role: "system", content: getPersonalitySystemPrompt(activeSession.personality_id)});
const summary = buildContextSummaryPrompt(contextSummaryOverride === undefined
? (upToMessageId == null ? activeSession.context_summary : null) : contextSummaryOverride);
if (summary) messagesPayload.push({role: "system", content: summary});
const memory = buildMemoryPrompt(repositories.listChatMemories(chatId, user));
if (memory) messagesPayload.push({role: "system", content: memory});
const personaState = buildPersonaStatePrompt(repositories.getPersonaChatState(chatId, user));
if (personaState) messagesPayload.push({
role: "system",
content: `Conversation continuity notes (context only):\n${personaState}`
});
const conversationMessages = conversation.map(entry => ({
role: entry.role === "bot" ? "assistant" : "user",
content: entry.content
}));
messagesPayload.push(...conversationMessages.slice(0, -1));
const currentQuestion = conversation.at(-1)?.content || "";
const selectedImages = visionSelection || attachmentService?.selectImages(
message == null ? history : [...history, {role: "user", content: message}], currentQuestion);
const visionContext = attachmentService?.context(selectedImages, currentQuestion);
if (visionContext) messagesPayload.push(...visionContext);
if (webSearchContext) messagesPayload.push(webSearchContext[0]);
if (webSearchContext?.[1]) messagesPayload.push(webSearchContext[1]);
if (conversationMessages.length) messagesPayload.push(conversationMessages.at(-1));
return {session: activeSession, activePersona, activeUserPersona, messagesPayload};
};
}