import fs from 'node:fs'; import crypto from 'node:crypto'; import path from 'node:path'; import { Readable, Transform, Writable } from 'node:stream'; import { pipeline } from 'node:stream/promises'; import { Agent, fetch as undiciFetch } from 'undici'; import { resolveBillingTokenState } from './billing-token-state.mjs'; import { appendBalanceEvent, createSseBillingTransform } from './sse-billing.mjs'; import { finalizeSessionStreamEvent, writeSseErrorAndEnd } from './sse-event-taxonomy.mjs'; import { developerToolsFromPolicy } from './capabilities.mjs'; import { evaluateProxyRequest, isNativeH5ApiPath } from './policies.mjs'; import { buildPublicUrl, buildSandboxSessionConstraints, PUBLISH_ROOT_DIR, resolvePublicBaseUrl, } from './user-publish.mjs'; import { buildCurrentTimeAgentPrefix, buildTaskRoutingAgentText } from './user-memory-profile.mjs'; import { reconcileAgentSession } from './session-reconcile.mjs'; import { consumeSessionEventsUntilFinish } from './session-reply-wait.mjs'; import { createSessionAccess, isSessionBrokerEnabled } from './session-broker.mjs'; import { createSessionBrokerMetrics, isSessionBrokerMetricsEnabled } from './session-broker-metrics.mjs'; import { createImgproxySigner } from './imgproxy-signer.mjs'; import { isDirectChatSessionId } from './direct-chat-service.mjs'; import { ensureGooseUserMessageMetadata } from './goose-message.mjs'; import { filterMemoriesByQuery, resolveMemoriesWithLegacyFallback } from './memory-legacy-fallback.mjs'; import { collectPageDataDeliveryArtifacts, rewritePageDataDeliveryLinks, } from './mindspace-page-data-finish-guard.mjs'; import { extractAttachmentText } from './mindspace-attachment-text.mjs'; import { buildCurrentTurnImageScopeNote, extractCurrentTurnImageUrls, scrubConversationHistoricalImageAttachments, } from './chat-image-turn-scope.mjs'; import { repairConversationToolHistory } from './chat-tool-history-repair.mjs'; import { buildVisionThumbnailBuffer } from './vision-image-thumb.mjs'; import { memoryLimitForIntervention, resolveMemoryInterventionMode, } from './memory-intervention.mjs'; import { formatSessionStreamSseChunk, isSessionStreamReplayEnabled, parseSessionSseBlock, parseSessionStreamLastEventId, resolveUpstreamResumeEventId, shouldPersistSessionStreamEvent, shouldSkipUpstreamAfterSessionReplay, } from './session-stream.mjs'; const insecureDispatcher = new Agent({ connect: { rejectUnauthorized: false }, }); function isHttpsTarget(target) { return target.startsWith('https://'); } function sanitizeUserFacingProxyMessage(message, fallback = '后端连接失败，请稍后重试') { const normalized = String(message ?? '').trim(); if (!normalized) return fallback; if (!/goose|goosed/i.test(normalized)) return normalized; if (/超时|timeout/i.test(normalized)) { return '后端连接超时，请确认后端服务正常后重试'; } if (/不可用|连接失败|failed to fetch|networkerror|fetch failed|upstream|econn|enotfound/i.test(normalized)) { return fallback; } return normalized .replace(/\bgoosed\b/gi, '后端服务') .replace(/\bgoose\b/gi, '后端'); } const PUBLIC_IMAGE_STORAGE_KEY_PATTERN = /^users\/([^/]+)\/images\/(\d{4}-\d{2}-\d{2}\/[^?#"'<>@\s]+)$/i; function extractPublicImageStorageKey(rawUrl) { const value = String(rawUrl ?? '').trim(); if (!value) return null; const directMatch = value.match(/(^|\/)(users\/[^?#"'<>@\s]+\/images\/\d{4}-\d{2}-\d{2}\/[^?#"'<>@\s]+)/i); if (directMatch?.[2]) return directMatch[2]; const plainIndex = value.indexOf('plain/local:///'); if (plainIndex >= 0) { const tail = value.slice(plainIndex + 'plain/local:///'.length); const storageKey = tail.replace(/@[a-z0-9]+(?:[?#].*)?$/i, '').replace(/[?#].*$/, ''); return storageKey || null; } return null; } function resolvePublicBaseUrlFromLayout(publishLayout) { const publicUrl = String(publishLayout?.publicUrl ?? '').trim(); if (!publicUrl) return resolvePublicBaseUrl(); try { return new URL(publicUrl).origin; } catch { return resolvePublicBaseUrl(); } } function buildPublicStandardImageUrl(rawUrl, userId, publishLayout = null) { const storageKey = extractPublicImageStorageKey(rawUrl); if (!storageKey) return null; const match = storageKey.match(PUBLIC_IMAGE_STORAGE_KEY_PATTERN); if (!match || match[1] !== String(userId ?? '')) return null; return buildPublicUrl(resolvePublicBaseUrlFromLayout(publishLayout), userId, `public/images/${match[2]}`); } async function apiFetch(target, apiSecret, pathname, init = {}) { const url = new URL(pathname, target); const headers = { ...(init.headers ?? {}), 'X-Secret-Key': apiSecret, }; if (init.body && !headers['Content-Type']) { headers['Content-Type'] = 'application/json'; } return undiciFetch(url, { ...init, headers, dispatcher: isHttpsTarget(target) ? insecureDispatcher : undefined, }); } function createRuntimeRouter({ targets, targetHealthy, redisUrl = process.env.MEMIND_RUNTIME_REDIS_URL, namespace = process.env.MEMIND_RUNTIME_REDIS_NAMESPACE ?? 'memind:runtime', }) { if (!redisUrl || targets.length <= 1) return null; let clientPromise = null; let warned = false; const workerIdForTarget = (target) => { const idx = targets.indexOf(target); return idx >= 0 ? `goosed-${idx + 1}` : `goosed-${Buffer.from(String(target)).toString('base64url')}`; }; const key = (...parts) => [namespace, ...parts].join(':'); const getClient = async () => { if (!clientPromise) { clientPromise = import('redis') .then(async ({ createClient }) => { const client = createClient({ url: redisUrl }); client.on('error', (err) => { if (!warned) { warned = true; console.warn('[RuntimeRouter] Redis error:', err instanceof Error ? err.message : err); } }); await client.connect(); console.log('[RuntimeRouter] Redis scheduler enabled'); return client; }) .catch((err) => { clientPromise = null; if (!warned) { warned = true; console.warn('[RuntimeRouter] Redis disabled:', err instanceof Error ? err.message : err); } return null; }); } return clientPromise; }; const readNumber = (value) => { const n = Number(value ?? 0); return Number.isFinite(n) ? n : 0; }; const parseFirstTokenSample = (value) => { const parts = String(value ?? '').split(':'); return readNumber(parts[1]); }; const percentile = (values, pct) => { const sorted = values .map((value) => readNumber(value)) .filter((value) => value > 0) .sort((a, b) => a - b); if (sorted.length === 0) return 0; const idx = Math.min(sorted.length - 1, Math.max(0, Math.ceil((pct / 100) * sorted.length) - 1)); return sorted[idx]; }; const firstTokenWindowStats = async (client, workerId, windowMs) => { if (!client) return { count: 0, p50Ms: 0, p95Ms: 0 }; const now = Date.now(); const samples = await client .sendCommand([ 'ZRANGEBYSCORE', key('worker', workerId, 'first_token_samples'), String(now - windowMs), String(now), ]) .catch(() => []); const values = samples.map(parseFirstTokenSample).filter((value) => value > 0); return { count: values.length, p50Ms: percentile(values, 50), p95Ms: percentile(values, 95), }; }; const scoreWorker = async (client, target) => { const workerId = workerIdForTarget(target); const values = await client.mGet([ key('worker', workerId, 'active_streams'), key('worker', workerId, 'active_sessions'), key('worker', workerId, 'ewma_first_token_ms'), key('worker', workerId, 'error_rate'), key('worker', workerId, 'memory_pressure'), key('worker', workerId, 'cpu_load'), key('worker', workerId, 'fd_pressure'), key('worker', workerId, 'drain'), ]); if (/^(1|true|yes)$/i.test(String(values[7] ?? ''))) return Number.POSITIVE_INFINITY; return ( readNumber(values[0]) * 3 + readNumber(values[1]) + readNumber(values[2]) * 0.01 + readNumber(values[3]) * 5 + readNumber(values[4]) * 2 + readNumber(values[5]) * 2 + readNumber(values[6]) * 2 ); }; const workerScoreFromValues = (values = []) => ( readNumber(values[0]) * 3 + readNumber(values[1]) + readNumber(values[2]) * 0.01 + readNumber(values[3]) * 5 + readNumber(values[4]) * 2 + readNumber(values[12]) * 2 + readNumber(values[13]) * 2 ); return { async pickTarget(fallbackPick, orderedTargets = targets) { const client = await getClient(); if (!client) return fallbackPick(); let winner = null; for (const target of orderedTargets) { if (!await targetHealthy(target)) continue; const score = await scoreWorker(client, target).catch(() => Number.POSITIVE_INFINITY); if (!Number.isFinite(score)) continue; if (!winner || score < winner.score) winner = { target, score }; } return winner?.target ?? fallbackPick(); }, async registerSession(sessionId, target) { const client = await getClient(); if (!client || !sessionId || !target) return; const workerId = workerIdForTarget(target); await client .multi() .set(key('session', sessionId, 'worker'), workerId, { EX: 60 * 60 * 24 * 30 }) .set(key('session', sessionId, 'target'), target, { EX: 60 * 60 * 24 * 30 }) .exec() .catch(() => null); }, async resolveSessionTarget(sessionId) { const client = await getClient(); if (!client || !sessionId) return null; const target = await client.get(key('session', sessionId, 'target')).catch(() => null); return target && targets.includes(target) ? target : null; }, async streamStarted(sessionId, target) { const client = await getClient(); if (!client || !target) return; const workerId = workerIdForTarget(target); const streamKey = sessionId ? key('stream', sessionId, 'status') : null; const now = String(Date.now()); const multi = client .multi() .incr(key('worker', workerId, 'active_streams')) .incr(key('worker', workerId, 'stream_open_count')) .set(key('worker', workerId, 'last_stream_started_at'), now) .set(key('worker', workerId, 'heartbeat'), now, { EX: 30 }); if (streamKey) multi.set(streamKey, 'active', { EX: 60 * 60 }); await multi.exec().catch(() => null); }, async streamEnded(sessionId, target, { status = 'closed' } = {}) { const client = await getClient(); if (!client || !target) return; const workerId = workerIdForTarget(target); const streamKey = sessionId ? key('stream', sessionId, 'status') : null; const activeKey = key('worker', workerId, 'active_streams'); const now = String(Date.now()); const nextValue = await client.decr(activeKey).catch(() => null); const multi = client .multi() .set(key('worker', workerId, 'last_stream_ended_at'), now) .set(key('worker', workerId, 'heartbeat'), now, { EX: 30 }); if (Number(nextValue ?? 0) < 0) multi.set(activeKey, '0'); if (status === 'aborted') multi.incr(key('worker', workerId, 'stream_abort_count')); if (status === 'error') multi.incr(key('worker', workerId, 'stream_error_count')); if (streamKey) multi.set(streamKey, status, { EX: 600 }); await multi.exec().catch(() => null); }, async firstTokenObserved(target, latencyMs) { const client = await getClient(); if (!client || !target) return; const workerId = workerIdForTarget(target); const sample = Math.max(0, Math.round(Number(latencyMs) || 0)); const ewmaKey = key('worker', workerId, 'ewma_first_token_ms'); const samplesKey = key('worker', workerId, 'first_token_samples'); const prev = readNumber(await client.get(ewmaKey).catch(() => null)); const next = prev > 0 ? Math.round(prev * 0.8 + sample * 0.2) : sample; const nowMs = Date.now(); const now = String(nowMs); await client .multi() .set(ewmaKey, String(next)) .set(key('worker', workerId, 'last_first_token_ms'), String(sample)) .set(key('worker', workerId, 'last_first_token_at'), now) .incr(key('worker', workerId, 'first_token_count')) .set(key('worker', workerId, 'heartbeat'), now, { EX: 30 }) .exec() .catch(() => null); await client .sendCommand([ 'ZADD', samplesKey, String(nowMs), `${now}:${sample}:${Math.random().toString(36).slice(2)}`, ]) .catch(() => null); await client .sendCommand(['ZREMRANGEBYSCORE', samplesKey, '0', String(nowMs - 60 * 60 * 1000)]) .catch(() => null); await client.sendCommand(['EXPIRE', samplesKey, String(2 * 60 * 60)]).catch(() => null); }, async getStatus() { const client = await getClient(); const workers = []; for (const target of targets) { const workerId = workerIdForTarget(target); const values = client ? await client .mGet([ key('worker', workerId, 'active_streams'), key('worker', workerId, 'active_sessions'), key('worker', workerId, 'ewma_first_token_ms'), key('worker', workerId, 'error_rate'), key('worker', workerId, 'memory_pressure'), key('worker', workerId, 'heartbeat'), key('worker', workerId, 'drain'), key('worker', workerId, 'stream_open_count'), key('worker', workerId, 'stream_abort_count'), key('worker', workerId, 'stream_error_count'), key('worker', workerId, 'last_stream_started_at'), key('worker', workerId, 'last_stream_ended_at'), key('worker', workerId, 'cpu_load'), key('worker', workerId, 'fd_pressure'), key('worker', workerId, 'fd_count'), key('worker', workerId, 'container_pids'), key('worker', workerId, 'container_health'), key('worker', workerId, 'metrics_sampled_at'), key('worker', workerId, 'last_first_token_ms'), key('worker', workerId, 'last_first_token_at'), key('worker', workerId, 'first_token_count'), key('worker', workerId, 'heartbeat_source'), key('worker', workerId, 'heartbeat_ok'), key('worker', workerId, 'heartbeat_status_code'), key('worker', workerId, 'heartbeat_latency_ms'), key('worker', workerId, 'heartbeat_error'), ]) .catch(() => []) : []; const firstToken5m = await firstTokenWindowStats(client, workerId, 5 * 60 * 1000); const firstToken1h = await firstTokenWindowStats(client, workerId, 60 * 60 * 1000); workers.push({ id: workerId, target, activeStreams: readNumber(values?.[0]), activeSessions: readNumber(values?.[1]), ewmaFirstTokenMs: readNumber(values?.[2]), errorRate: readNumber(values?.[3]), memoryPressure: readNumber(values?.[4]), heartbeat: values?.[5] ? Number(values[5]) : null, drain: /^(1|true|yes)$/i.test(String(values?.[6] ?? '')), streamOpenCount: readNumber(values?.[7]), streamAbortCount: readNumber(values?.[8]), streamErrorCount: readNumber(values?.[9]), lastStreamStartedAt: values?.[10] ? Number(values[10]) : null, lastStreamEndedAt: values?.[11] ? Number(values[11]) : null, cpuLoad: readNumber(values?.[12]), fdPressure: readNumber(values?.[13]), fdCount: readNumber(values?.[14]), containerPids: readNumber(values?.[15]), containerHealth: values?.[16] ?? null, metricsSampledAt: values?.[17] ? Number(values[17]) : null, lastFirstTokenMs: readNumber(values?.[18]), lastFirstTokenAt: values?.[19] ? Number(values[19]) : null, firstTokenCount: readNumber(values?.[20]), heartbeatSource: values?.[21] ?? null, heartbeatOk: values?.[22] == null ? null : /^(1|true|yes)$/i.test(String(values[22])), heartbeatStatusCode: readNumber(values?.[23]), heartbeatLatencyMs: readNumber(values?.[24]), heartbeatError: values?.[25] || null, firstToken5m, firstToken1h, score: workerScoreFromValues(values), }); } return { enabled: Boolean(client), namespace, workers, }; }, }; } async function readJsonBody(req) { if (req.method === 'GET' || req.method === 'HEAD') return null; const chunks = []; for await (const chunk of req) { chunks.push(chunk); } if (chunks.length === 0) return null; const raw = Buffer.concat(chunks).toString('utf8'); if (!raw.trim()) return null; return JSON.parse(raw); } function sendProxyResponse(res, upstream) { res.status(upstream.status); upstream.headers.forEach((value, key) => { if (key === 'transfer-encoding') return; res.setHeader(key, value); }); if (!upstream.body) { res.end(); return; } Readable.fromWeb(upstream.body).pipe(res); } function writeSseProxyErrorAndEnd(res, message) { writeSseErrorAndEnd(res, message, { status: 502 }); } function extractSessionId(req, body) { const fromParams = req.params?.sessionId ?? req.params?.id; if (fromParams) return fromParams; if (body?.session_id) return body.session_id; if (body?.sessionId) return body.sessionId; return null; } function firstUserText(message) { return ( message?.content?.find?.((item) => item?.type === 'text' && typeof item.text === 'string')?.text ?? '' ); } const IMAGE_URL_LINE_RE = /^\[图片\d+]:\s*(\S.+)$/; const FILE_ATTACHMENT_LINE_RE = /^\[文件\d+: ([^\]]+)\]: (.+)$/; const PUBLIC_HTML_LINK_PATTERN = /https?:\/\/[^\s<>"')\]]+\/MindSpace\/([0-9a-f-]{36}|[a-z0-9._-]+)\/([^\s<>"')\]]+\.html)/gi; const PUBLIC_HTML_MARKDOWN_LINK_PATTERN = /\[([^\]\n]*)\]\((https?:\/\/[^\s<>"')\]]+\/MindSpace\/([0-9a-f-]{36}|[a-z0-9._-]+)\/([^\s<>"')\]]+\.html))\)/gi; const MARKDOWN_LINK_PATTERN = /\[([^\]\n]*)\]\((https?:\/\/[^\s<>"')\]]+)\)/g; const PLAIN_HTML_URL_PATTERN = /https?:\/\/[^\s<>"')\]]+\.html/gi; const INLINE_PUBLIC_HTML_PATH_PATTERN = /`((?:\.\/)?public\/[^`\s<>"')\]]+\.html)`/gi; const MISSING_PUBLIC_HTML_NOTICE_PATTERN = /（页面生成未完成，已阻止显示失效链接：([^（）\n]+\.html)。）/g; const USER_IDENTITY_BLOCK_PATTERN = /^\[用户身份\][\s\S]*?(?:\n{2,}|$)/; const IMAGE_URL_LINES_PATTERN = /\n*\[图片\d+]: [^\n]+/g; const TKMIND_VISION_NOTE_PATTERN = /\n*【TKMind 图片分析结果[\s\S]*$/; function decodePathSegment(segment) { try { return decodeURIComponent(segment); } catch { return segment; } } function encodeUrlPath(relativePath) { return String(relativePath ?? '') .split('/') .filter(Boolean) .map((part) => encodeURIComponent(part)) .join('/'); } function normalizeStaticHtmlRelativePath(relativePath) { const parts = String(relativePath ?? '') .replace(/^\/+/, '') .split('/') .filter((part) => part && part !== '.' && part !== '..'); if (parts.length === 0) return ''; if (parts[0].toLowerCase() === 'public') return ['public', ...parts.slice(1)].join('/'); if (parts.length === 1 && parts[0].toLowerCase().endsWith('.html')) return `public/${parts[0]}`; return parts.join('/'); } function canonicalizeStaticPageUrl(publicUrl, originalRelativePath, canonicalRelativePath) { const originalClean = String(originalRelativePath ?? '').replace(/^\/+/, ''); if (!canonicalRelativePath || canonicalRelativePath === originalClean) return publicUrl; const suffix = encodeUrlPath(originalClean).replace(/[.*+?^${}()|[\]\\]/g, '\\$&'); return String(publicUrl).replace(new RegExp(`${suffix}$`), encodeUrlPath(canonicalRelativePath)); } function buildMissingPublicHtmlNotice(filename) { return `（页面生成未完成，已阻止显示失效链接：${filename || '页面'}。）`; } function publicHtmlRootsForUser(owner) { const roots = [ path.resolve(process.cwd(), PUBLISH_ROOT_DIR, owner), ]; const sharedPublishRoots = [ process.env.GOOSED_SANDBOX_PUBLISH_ROOT, process.env.MEMIND_SHARED_PUBLISH_ROOT, ]; for (const configuredRoot of sharedPublishRoots) { const value = String(configuredRoot ?? '').trim(); if (value) roots.push(path.resolve(value, owner)); } const usersRoot = String(process.env.H5_USERS_ROOT ?? '').trim(); if (usersRoot) { roots.push(path.resolve(path.dirname(usersRoot), PUBLISH_ROOT_DIR, owner)); } return [...new Set(roots)]; } function publicHtmlExistsForUser(owner, relativePath, currentUser) { const normalizedOwner = String(owner ?? '').trim().toLowerCase(); const normalizedUserId = String(currentUser?.id ?? '').trim().toLowerCase(); const normalizedUsername = String(currentUser?.username ?? '').trim().toLowerCase(); if (!normalizedOwner) return false; if (normalizedOwner !== normalizedUserId && normalizedOwner !== normalizedUsername) return true; const normalizedRelativePath = normalizeStaticHtmlRelativePath(relativePath); if (!normalizedRelativePath || !normalizedRelativePath.toLowerCase().endsWith('.html')) return false; for (const root of publicHtmlRootsForUser(normalizedOwner)) { const target = path.resolve(root, normalizedRelativePath); if (target !== root && !target.startsWith(`${root}${path.sep}`)) continue; if (fs.existsSync(target) && fs.statSync(target).isFile()) return true; } return false; } function sanitizeOwnPublicHtmlUrl(publicUrl, owner, rawRelativePath, currentUser) { const relativePath = decodePathSegment(rawRelativePath); const canonicalRelativePath = normalizeStaticHtmlRelativePath(relativePath); if (!publicHtmlExistsForUser(owner, canonicalRelativePath, currentUser)) { return { ok: false, notice: buildMissingPublicHtmlNotice(path.posix.basename(canonicalRelativePath || relativePath)), }; } return { ok: true, url: buildPublicUrl(resolvePublicBaseUrl(), owner, canonicalRelativePath), }; } function isCurrentUserPublicOwner(owner, currentUser) { const normalizedOwner = String(owner ?? '').trim().toLowerCase(); const normalizedUserId = String(currentUser?.id ?? '').trim().toLowerCase(); const normalizedUsername = String(currentUser?.username ?? '').trim().toLowerCase(); return Boolean(normalizedOwner && ( normalizedOwner === normalizedUserId || normalizedOwner === normalizedUsername )); } function sanitizeMissingMindSpacePublicHtmlUrl(url, currentUser) { let parsed; try { parsed = new URL(url); } catch { return null; } const parts = parsed.pathname .split('/') .filter(Boolean) .map((part) => decodePathSegment(part)); if (parts[0]?.toLowerCase() === 'mindspace') return null; const owner = parts[0]; if (!isCurrentUserPublicOwner(owner, currentUser)) return null; const rawRelativePath = parts.slice(1).join('/'); if (!rawRelativePath.toLowerCase().endsWith('.html')) return null; const result = sanitizeOwnPublicHtmlUrl(url, owner, rawRelativePath, currentUser); return result.ok ? result.url : result.notice; } function sanitizeOwnPublicHtmlRelativePath(rawRelativePath, currentUser) { const owner = currentUser?.id || currentUser?.username; if (!owner) return null; const normalizedRelativePath = normalizeStaticHtmlRelativePath(rawRelativePath); if (!normalizedRelativePath.toLowerCase().endsWith('.html')) return null; if (!publicHtmlExistsForUser(owner, normalizedRelativePath, currentUser)) return null; return { label: path.posix.basename(normalizedRelativePath), url: buildPublicUrl(resolvePublicBaseUrl(), owner, normalizedRelativePath), }; } function rewriteOwnPublicationRouteLinks(text, currentUser) { const owner = String(currentUser?.id ?? currentUser?.username ?? '').trim(); if (!owner) return String(text ?? ''); const publishDir = path.resolve(process.cwd(), PUBLISH_ROOT_DIR, owner); const artifacts = collectPageDataDeliveryArtifacts(publishDir); if (!artifacts.length) return String(text ?? ''); return rewritePageDataDeliveryLinks(text, artifacts); } export function sanitizePublicHtmlLinksInText(text, currentUser) { let next = String(text ?? '').replace( MISSING_PUBLIC_HTML_NOTICE_PATTERN, (match, filename) => { const result = sanitizeOwnPublicHtmlRelativePath(`public/${path.posix.basename(filename)}`, currentUser); return result ? `[${result.label}](${result.url})` : match; }, ); next = next.replace( PUBLIC_HTML_MARKDOWN_LINK_PATTERN, (match, label, url, owner, rawRelativePath) => { const result = sanitizeOwnPublicHtmlUrl(url, owner, rawRelativePath, currentUser); if (!result.ok) return result.notice; return label ? `[${label}](${result.url})` : result.url; }, ); next = next.replace(PUBLIC_HTML_LINK_PATTERN, (match, owner, rawRelativePath) => { const result = sanitizeOwnPublicHtmlUrl(match, owner, rawRelativePath, currentUser); return result.ok ? result.url : result.notice; }); next = next.replace(MARKDOWN_LINK_PATTERN, (match, label, url) => { const sanitizedUrl = sanitizeMissingMindSpacePublicHtmlUrl(url, currentUser); if (!sanitizedUrl) return match; return label ? `[${label}](${sanitizedUrl})` : sanitizedUrl; }); next = next.replace(PLAIN_HTML_URL_PATTERN, (match) => ( sanitizeMissingMindSpacePublicHtmlUrl(match, currentUser) ?? match )); next = next.replace(INLINE_PUBLIC_HTML_PATH_PATTERN, (match, rawRelativePath) => { const result = sanitizeOwnPublicHtmlRelativePath(rawRelativePath, currentUser); return result ? `[${result.label}](${result.url})` : match; }); return rewriteOwnPublicationRouteLinks(next, currentUser); } export function sanitizeUserVisibleMessageText(text, currentUser) { return sanitizePublicHtmlLinksInText( String(text ?? '') .replace(USER_IDENTITY_BLOCK_PATTERN, '') .replace(TKMIND_VISION_NOTE_PATTERN, '') .replace(IMAGE_URL_LINES_PATTERN, '') .trim(), currentUser, ); } export function sanitizeSessionMessagePublicHtmlLinks(message, currentUser) { if (!message || !Array.isArray(message.content)) return message; let changed = false; const imageUrls = extractImageUrlsFromMessage(message); const content = message.content.map((item) => { if (item?.type !== 'text' || typeof item.text !== 'string') return item; const nextText = sanitizeUserVisibleMessageText(item.text, currentUser); if (nextText === item.text) return item; changed = true; return { ...item, text: nextText }; }); const currentDisplayText = typeof message.metadata?.displayText === 'string' ? message.metadata.displayText : null; const nextDisplayText = currentDisplayText == null ? null : sanitizeUserVisibleMessageText(currentDisplayText, currentUser); const metadataChanged = (currentDisplayText != null && nextDisplayText !== currentDisplayText) || (!Array.isArray(message.metadata?.imageUrls) && imageUrls.length > 0); if (!changed && !metadataChanged) return message; return { ...message, content, metadata: { ...(message.metadata ?? {}), ...(nextDisplayText != null ? { displayText: nextDisplayText } : {}), ...(!Array.isArray(message.metadata?.imageUrls) && imageUrls.length > 0 ? { imageUrls } : {}), }, }; } export function sanitizeSessionConversationPublicHtmlLinks(conversation, currentUser) { if (!Array.isArray(conversation)) return conversation; return conversation.map((message) => sanitizeSessionMessagePublicHtmlLinks(message, currentUser)); } function createSessionEventSanitizer(currentUser, { onEvent, onPersistFrame, normalizeReplayIds = false } = {}) { let buffer = ''; const flushChunk = (controller, block) => { if (!block) return; const parsed = parseSessionSseBlock(block); if (!parsed.data) { controller.push(block); return; } let event; try { event = JSON.parse(parsed.data); } catch { controller.push(block); return; } event = finalizeSessionStreamEvent(event); if (typeof onEvent === 'function') { try { onEvent(event); } catch { // Ignore side-effect failures and keep SSE flowing to the client. } } if (event?.type === 'Message' && event.message) { event.message = sanitizeSessionMessagePublicHtmlLinks(event.message, currentUser); } else if (event?.type === 'UpdateConversation' && Array.isArray(event.conversation)) { event.conversation = sanitizeSessionConversationPublicHtmlLinks(event.conversation, currentUser); } if (normalizeReplayIds) { const frameId = parsed.id || crypto.randomUUID(); if (typeof onPersistFrame === 'function' && shouldPersistSessionStreamEvent(event)) { try { onPersistFrame({ id: frameId, upstreamEventId: parsed.id ?? null, payload: event }); } catch { // Ignore persistence failures and keep SSE flowing to the client. } } controller.push(formatSessionStreamSseChunk({ id: frameId, event: parsed.eventName, data: event, })); return; } const lines = block.split('\n'); const sanitizedLines = lines.map((line) => { if (!line.startsWith('data: ')) return line; return `data: ${JSON.stringify(event)}`; }); controller.push(`${sanitizedLines.join('\n')}\n`); }; return new Transform({ transform(chunk, _encoding, callback) { buffer += chunk.toString('utf8'); let boundary = buffer.indexOf('\n\n'); while (boundary >= 0) { const block = buffer.slice(0, boundary + 2); buffer = buffer.slice(boundary + 2); flushChunk(this, block); boundary = buffer.indexOf('\n\n'); } callback(); }, flush(callback) { flushChunk(this, buffer); buffer = ''; callback(); }, }); } export function extractImageUrlsFromMessage(userMessage) { return extractCurrentTurnImageUrls(userMessage); } function messageHasImages(userMessage) { return extractImageUrlsFromMessage(userMessage).length > 0; } export function extractFileAttachmentsFromMessage(userMessage) { const fromMetadata = Array.isArray(userMessage?.metadata?.fileAttachments) ? userMessage.metadata.fileAttachments .filter((item) => item?.downloadUrl?.trim() && item?.filename?.trim()) .map((item) => ({ assetId: String(item.assetId ?? '').trim(), filename: String(item.filename).trim(), downloadUrl: String(item.downloadUrl).trim(), mimeType: String(item.mimeType ?? 'application/octet-stream').trim(), })) : []; if (fromMetadata.length > 0) return fromMetadata; const parsed = []; const content = userMessage?.content; if (!Array.isArray(content)) return parsed; for (const item of content) { if (item?.type !== 'text' || typeof item.text !== 'string') continue; for (const line of item.text.split('\n')) { const match = line.trim().match(FILE_ATTACHMENT_LINE_RE); if (!match?.[1] || !match?.[2]) continue; parsed.push({ assetId: '', filename: match[1], downloadUrl: match[2], mimeType: 'application/octet-stream', }); } } return parsed; } function messageHasFileAttachments(userMessage) { return extractFileAttachmentsFromMessage(userMessage).length > 0; } function prependAgentTextToUserMessage(body, transformText) { const originalText = firstUserText(body?.user_message); if (!originalText?.trim()) return body; const agentText = transformText(originalText); if (!agentText || agentText === originalText) return body; const userMessage = body.user_message ?? {}; const content = Array.isArray(userMessage.content) ? [...userMessage.content] : []; const firstTextIndex = content.findIndex( (item) => item?.type === 'text' && typeof item.text === 'string', ); if (firstTextIndex >= 0) { content[firstTextIndex] = { ...content[firstTextIndex], text: agentText }; } else { content.unshift({ type: 'text', text: agentText }); } return { ...body, user_message: { ...userMessage, content, metadata: { ...(userMessage.metadata ?? {}), displayText: userMessage.metadata?.displayText && String(userMessage.metadata.displayText).trim() ? userMessage.metadata.displayText : originalText, }, }, }; } function injectCurrentTimeAnchor(body, timezone) { const tz = String(timezone ?? process.env.H5_DEFAULT_TIMEZONE ?? 'Asia/Shanghai').trim() || 'Asia/Shanghai'; return prependAgentTextToUserMessage(body, (originalText) => { if (originalText.includes('TKMind 当前时间基准')) return originalText; return `${buildCurrentTimeAgentPrefix({ timezone: tz })}${originalText}`; }); } function injectTaskRoutingHint(body, sessionPolicy) { return prependAgentTextToUserMessage(body, (originalText) => buildTaskRoutingAgentText(originalText, sessionPolicy), ); } export async function buildVisionPayload({ userMessage, userId, publishLayout, localFetchAsset, llmProviderService, imgproxySigner = null, fetchImpl = undiciFetch, }) { void imgproxySigner; if (!llmProviderService || !userId) return null; const rawImageUrls = extractCurrentTurnImageUrls(userMessage); if (rawImageUrls.length === 0) return null; const originalDisplayText = typeof userMessage?.metadata?.displayText === 'string' && userMessage.metadata.displayText.trim() ? userMessage.metadata.displayText : Array.isArray(userMessage?.content) ? userMessage.content .filter((item) => item?.type === 'text' && typeof item.text === 'string') .map((item) => item.text) .join('\n') .trim() : ''; const imageItems = []; for (const rawUrl of rawImageUrls) { try { const match = rawUrl.match(/\/mindspace\/v1\/assets\/([^/?#]+)(?:\/download)?/); let buffer; let mimeType = 'image/jpeg'; const fetchableUrl = (() => { if (/^https?:\/\//i.test(rawUrl)) return rawUrl; if (rawUrl.startsWith('/')) { const baseUrl = process.env.H5_PUBLIC_BASE_URL || 'http://127.0.0.1:8081'; return new URL(rawUrl, baseUrl).toString(); } return rawUrl; })(); try { const response = await fetchImpl(fetchableUrl); if (!response.ok) { throw new Error(`Vision fetch failed: ${response.status}`); } buffer = Buffer.from(await response.arrayBuffer()); mimeType = response.headers.get('content-type') || 'image/jpeg'; } catch (fetchErr) { if (!match || !localFetchAsset) { throw fetchErr; } const assetId = decodeURIComponent(match[1]); console.warn(`Vision fetch fallback for asset ${assetId}:`, fetchErr instanceof Error ? fetchErr.message : fetchErr); const asset = await localFetchAsset(userId, assetId); buffer = asset.buffer; mimeType = asset.mimeType; } let relativePath = rawUrl; try { const parsed = new URL(rawUrl); relativePath = parsed.pathname + parsed.search; } catch { /* keep rawUrl */ } const publicStandardUrl = buildPublicStandardImageUrl(rawUrl, userId, publishLayout); let visionBuffer = buffer; let visionMimeType = mimeType; try { visionBuffer = await buildVisionThumbnailBuffer(buffer, mimeType); visionMimeType = 'image/jpeg'; } catch (thumbErr) { console.warn( 'Vision thumbnail skipped:', thumbErr instanceof Error ? thumbErr.message : thumbErr, ); } imageItems.push({ mimeType, visionMimeType, data: visionBuffer.toString('base64'), relativePath, rawUrl, embedUrl: publicStandardUrl ?? relativePath, }); } catch (err) { console.warn('Vision image fetch skipped:', err instanceof Error ? err.message : err); } } if (imageItems.length === 0) return null; const visionDescription = await llmProviderService .analyzeImagesWithVision(imageItems, '请详细描述图片的视觉内容：主体、颜色、风格、构图，不要生成代码或页面方案') .catch(() => null); const pathList = imageItems .map((item, i) => `图片${i + 1}: 图片${i + 1}`) .join(' '); const publicUrlPrefix = publishLayout?.publicUrl ? `${String(publishLayout.publicUrl).replace(/\/$/, '')}/public/` : null; const injectedNote = `\n\n${buildCurrentTurnImageScopeNote(imageItems)}` + '【TKMind 图片分析结果 — 仅供执行参考，不要向用户复述此段内容】\n' + (visionDescription ? `Qwen VL 图片描述：\n${visionDescription}\n\n` : '') + '写作约束：不得改写图片里人物的年龄、性别、人数或主体关系；如果用户明确要求儿童语气或童趣风格，也只能调整表达方式，不能把图片主体改写成儿童场景。\n' + `图片 HTML 嵌入路径（直接写入 标签；以下都是无需 cookie 的公开压缩标准图片，禁止使用 local://、/users/ 私有路径、原图地址或需要登录态的下载链接）：\n${pathList}\n` + '执行要求：必须先调用 load_skill → static-page-publish（每次生成页面都要调用，不可省略），' + '再用 write_file 写入 public/页面.html（禁止用 shell/cat/heredoc 写 HTML，容器内文件不会出现在公网链接）；' + (publicUrlPrefix ? `完成后向用户给出 Markdown 可点击链接，格式：[页面标题](${publicUrlPrefix}<文件名>.html)；` : '完成后按技能说明里的链接格式给用户一个 Markdown 可点击链接；') + '不要向用户展示 HTML 代码块。'; let updatedContent = Array.isArray(userMessage?.content) ? [...userMessage.content] : []; for (const item of imageItems) { updatedContent = updatedContent.map((c) => { if (c?.type !== 'text' || typeof c.text !== 'string') return c; const updated = c.text.replaceAll(item.rawUrl, item.embedUrl); return updated === c.text ? c : { ...c, text: updated }; }); } const lastTextIdx = updatedContent.reduceRight( (found, item, i) => (found === -1 && item?.type === 'text' ? i : found), -1, ); if (lastTextIdx >= 0) { updatedContent[lastTextIdx] = { ...updatedContent[lastTextIdx], text: updatedContent[lastTextIdx].text + injectedNote, }; } else { updatedContent.push({ type: 'text', text: injectedNote }); } const canonicalImageUrls = imageItems .map((item) => item.embedUrl ?? item.rawUrl) .filter((url) => typeof url === 'string' && url.trim()); return { userMessage: { ...userMessage, content: updatedContent, metadata: { ...(userMessage.metadata ?? {}), ...(originalDisplayText ? { displayText: originalDisplayText } : {}), ...(canonicalImageUrls.length ? { imageUrls: canonicalImageUrls } : {}), }, }, billableImageCount: visionDescription ? 1 : 0, }; } export async function buildAttachmentPayload({ userMessage, userId, localFetchAsset, fetchImpl = undiciFetch, }) { if (!userId) return null; const attachments = extractFileAttachmentsFromMessage(userMessage); if (attachments.length === 0) return null; const originalDisplayText = typeof userMessage?.metadata?.displayText === 'string' && userMessage.metadata.displayText.trim() ? userMessage.metadata.displayText : Array.isArray(userMessage?.content) ? userMessage.content .filter((item) => item?.type === 'text' && typeof item.text === 'string') .map((item) => item.text) .join('\n') .trim() : ''; const sections = []; for (const [index, attachment] of attachments.entries()) { try { const match = attachment.downloadUrl.match(/\/mindspace\/v1\/assets\/([^/?#]+)(?:\/download)?/); let buffer; let mimeType = attachment.mimeType || 'application/octet-stream'; const fetchableUrl = (() => { if (/^https?:\/\//i.test(attachment.downloadUrl)) return attachment.downloadUrl; if (attachment.downloadUrl.startsWith('/')) { const baseUrl = process.env.H5_PUBLIC_BASE_URL || 'http://127.0.0.1:8081'; return new URL(attachment.downloadUrl, baseUrl).toString(); } return attachment.downloadUrl; })(); try { const response = await fetchImpl(fetchableUrl); if (!response.ok) { throw new Error(`Attachment fetch failed: ${response.status}`); } buffer = Buffer.from(await response.arrayBuffer()); mimeType = response.headers.get('content-type') || mimeType; } catch (fetchErr) { if (!match || !localFetchAsset) { throw fetchErr; } const assetId = decodeURIComponent(match[1]); const asset = await localFetchAsset(userId, assetId); buffer = asset.buffer; mimeType = asset.mimeType || mimeType; } const extracted = extractAttachmentText(buffer, mimeType, attachment.filename); const workspacePath = `oa/${attachment.filename}`; const isXlsx = /\.xlsx$/iu.test(attachment.filename); sections.push( [ `附件${index + 1}: ${attachment.filename}`, `下载链接: ${attachment.downloadUrl}`, `工作区路径: ${workspacePath}`, extracted.text ? `提取文本:\n${extracted.text.slice(0, 12000)}` : '提取文本: （未能自动提取正文，请用 read_file / mindspace_asset_download 读取源文件）', ...(isXlsx ? ['Excel 分析提示: 若当前会话提供 excel_inspect / excel_analyze / excel_chart / excel_report，必须优先使用这些工具读取完整工作簿；以上提取文本仅作预览。生成 HTML 分析报告时必须使用 excel_report 返回的已校验页面，禁止用 write_file / edit_file 手工抄写数字。'] : []), ...(extracted.warnings?.length ? [`提示: ${extracted.warnings.join(', ')}`] : []), ].join('\n'), ); } catch (err) { console.warn('Attachment analysis skipped:', err instanceof Error ? err.message : err); sections.push( `附件${index + 1}: ${attachment.filename}\n下载链接: ${attachment.downloadUrl}\n提取文本: （读取失败，请用 read_file oa/${attachment.filename} 或 mindspace_asset_download 读取）`, ); } } if (sections.length === 0) return null; const injectedNote = '\n\n【TKMind 附件分析结果 — 仅供执行参考，不要向用户复述此段内容】\n' + `${sections.join('\n\n')}\n` + '执行要求：普通附件优先基于上述提取文本回答；Excel 附件在专用工具可用时必须优先调用专用工具完成结构检查与分析。若需核对源文件，直接 read_file oa/文件名 或 mindspace_asset_download 落盘后分析，不要要求用户去界面重新上传。'; const updatedContent = Array.isArray(userMessage?.content) ? [...userMessage.content] : []; const lastTextIdx = updatedContent.reduceRight( (found, item, i) => (found === -1 && item?.type === 'text' ? i : found), -1, ); if (lastTextIdx >= 0) { updatedContent[lastTextIdx] = { ...updatedContent[lastTextIdx], text: updatedContent[lastTextIdx].text + injectedNote, }; } else { updatedContent.push({ type: 'text', text: injectedNote }); } return { userMessage: { ...userMessage, content: updatedContent, metadata: { ...(userMessage.metadata ?? {}), ...(originalDisplayText ? { displayText: originalDisplayText } : {}), }, }, }; } export function createTkmindProxy({ // GOOSED PROXY BOUNDARY: canonical H5 chat → goosed adapter (Patch 5, goosed-proxy-boundary.mjs) apiTarget, apiTargets, apiSecret, userAuth, sessionAccess = null, sessionStreamStore = null, llmProviderService, localFetchAsset, subscriptionService, sessionSnapshotService, conversationMemoryService, memoryV2, systemDisclosurePolicyService = null, }) { const sessionStore = sessionAccess ?? createSessionAccess({ userAuth, enabled: isSessionBrokerEnabled() }); const brokerMetrics = isSessionBrokerMetricsEnabled() && sessionStore.enabled ? createSessionBrokerMetrics({ logger: console }) : null; const targets = apiTargets?.length ? apiTargets : apiTarget ? [apiTarget] : []; const primaryTarget = targets[0] ?? apiTarget ?? ''; let rrIdx = 0; let imgproxySigner = null; if (process.env.IMGPROXY_BASE_URL && process.env.IMGPROXY_SIGNING_KEY && process.env.IMGPROXY_SIGNING_SALT) { try { imgproxySigner = createImgproxySigner( process.env.IMGPROXY_SIGNING_KEY, process.env.IMGPROXY_SIGNING_SALT, ); console.log('[TKMindProxy] imgproxy signer initialized'); } catch (err) { console.warn('[TKMindProxy] imgproxy signer init failed:', err instanceof Error ? err.message : err); } } async function targetHealthy(target) { try { const upstream = await apiFetch(target, apiSecret, '/status', { method: 'GET', signal: AbortSignal.timeout(1500), }); return upstream.ok; } catch { return false; } } const runtimeRouter = createRuntimeRouter({ targets, targetHealthy }); function isGeneratedSessionName(name) { const normalized = typeof name === 'string' ? name.trim() : ''; return /^20\d{6}(?:[_-]\d+)?$/.test(normalized); } // A session "name" that carries no real topic: empty, or Goose's default // placeholder before its describe step has run. function hasDefaultSessionName(session) { if (session?.user_set_name) return false; const name = typeof session?.name === 'string' ? session.name.trim() : ''; return name === '' || name === 'New Chat' || name === '新对话' || isGeneratedSessionName(name); } // For sessions still carrying a default name, fill in a title derived from the // first user message (kept in the snapshot cache) so the history list shows a // meaningful label instead of "会话 ". Best-effort: never blocks the list. async function enrichSessionHistory(sessions, userId) { const needsSummary = sessions.filter( (session) => hasDefaultSessionName(session) || Number(session?.message_count ?? 0) === 0, ); if (needsSummary.length === 0) return; if (sessionSnapshotService?.isEnabled?.()) { try { const summaries = await sessionSnapshotService.getHistorySummaries(needsSummary.map((s) => s.id)); for (const session of needsSummary) { const summary = summaries.get(session.id); if (!summary) continue; if (summary.title && hasDefaultSessionName(session)) { session.name = summary.title; } if (Number(session?.message_count ?? 0) === 0 && Number(summary.messageCount ?? 0) > 0) { session.message_count = Number(summary.messageCount); } } } catch { // Non-fatal: fall back to DB summaries below. } } const stillNeedsSummary = needsSummary.filter( (session) => hasDefaultSessionName(session) || Number(session?.message_count ?? 0) === 0, ); if (stillNeedsSummary.length === 0 || !conversationMemoryService?.loadSessionListSummaries) return; try { const dbSummaries = await conversationMemoryService.loadSessionListSummaries( userId, stillNeedsSummary.map((session) => session.id), ); for (const session of stillNeedsSummary) { const summary = dbSummaries.get(session.id); if (!summary) continue; if (summary.title && hasDefaultSessionName(session)) { session.name = summary.title; } if (Number(session?.message_count ?? 0) === 0 && Number(summary.messageCount ?? 0) > 0) { session.message_count = Number(summary.messageCount); } if (!session.updated_at && summary.updatedAt) { session.updated_at = summary.updatedAt; } } } catch { // Non-fatal: fall back to the client-side label. } } function projectSessionSummary(session) { return { id: session?.id, name: typeof session?.name === 'string' ? session.name : '', message_count: Number(session?.message_count ?? 0), created_at: session?.created_at ?? undefined, updated_at: session?.updated_at ?? undefined, user_set_name: Boolean(session?.user_set_name), recipe: session?.recipe ?? null, }; } function sortSessionsByRecent(left, right) { const leftTime = Date.parse(left?.updated_at ?? left?.created_at ?? '') || 0; const rightTime = Date.parse(right?.updated_at ?? right?.created_at ?? '') || 0; return rightTime - leftTime; } function matchesSessionQuery(summary, rawQuery) { const query = String(rawQuery ?? '').trim().toLowerCase(); if (!query) return true; const haystacks = [ summary?.name, summary?.recipe?.title, summary?.id, ] .filter((value) => typeof value === 'string' && value.trim()) .map((value) => value.toLowerCase()); return haystacks.some((value) => value.includes(query)); } function paginateSessionConversation(conversation, beforeValue, limitValue) { const visibleConversation = Array.isArray(conversation) ? conversation.filter((message) => message?.metadata?.userVisible) : []; const total = visibleConversation.length; const before = Math.max(Number(beforeValue ?? 0) || 0, 0); const requestedLimit = Number(limitValue ?? 0) || 0; if (requestedLimit <= 0) { return { conversation: visibleConversation, page: { total, before: 0, limit: total, returned: total, has_more_before: false, }, }; } const limit = Math.min(Math.max(requestedLimit, 1), 200); const end = Math.max(total - before, 0); const start = Math.max(end - limit, 0); const paged = visibleConversation.slice(start, end); return { conversation: paged, page: { total, before, limit, returned: paged.length, has_more_before: start > 0, }, }; } async function pickTarget() { if (targets.length <= 1) return primaryTarget; const fallbackPick = async () => { for (let i = 0; i < targets.length; i += 1) { const target = targets[rrIdx]; rrIdx = (rrIdx + 1) % targets.length; if (await targetHealthy(target)) return target; } return primaryTarget; }; if (runtimeRouter) { const orderedTargets = targets.map((_, i) => targets[(rrIdx + i) % targets.length]); rrIdx = (rrIdx + 1) % targets.length; return runtimeRouter.pickTarget(fallbackPick, orderedTargets); } return fallbackPick(); } async function rememberSessionTarget(sessionId, target) { if (!runtimeRouter || !sessionId || !target) return; await runtimeRouter.registerSession(sessionId, target).catch(() => null); } async function markStreamStarted(sessionId, target) { if (!runtimeRouter || !sessionId || !target) return; await runtimeRouter.streamStarted(sessionId, target).catch(() => null); } async function markStreamEnded(sessionId, target, options = {}) { if (!runtimeRouter || !sessionId || !target) return; await runtimeRouter.streamEnded(sessionId, target, options).catch(() => null); } function markFirstTokenObserved(target, latencyMs) { if (!runtimeRouter || !target) return; void runtimeRouter.firstTokenObserved(target, latencyMs).catch(() => null); } async function getRuntimeStatus() { const targetStatuses = []; for (const target of targets) { targetStatuses.push({ target, healthy: await targetHealthy(target), }); } const routerStatus = runtimeRouter ? await runtimeRouter.getStatus().catch((err) => ({ enabled: false, error: err instanceof Error ? err.message : String(err), })) : { enabled: false, reason: 'MEMIND_RUNTIME_REDIS_URL not configured' }; return { publicBaseUrl: process.env.H5_PUBLIC_BASE_URL ?? null, router: routerStatus, toolRuntime: { defaultMode: 'chat', codeToolMode: 'code', chatInjectsCodeTools: false, codeRunsEnabled: ['1', 'true', 'yes', 'on'].includes( String(process.env.MEMIND_AGENT_CODE_RUNS_ENABLED ?? '').trim().toLowerCase(), ), aiderTimeoutMs: Number(process.env.MEMIND_AIDER_TIMEOUT_MS ?? 600_000), openhandsTimeoutMs: Number(process.env.MEMIND_OPENHANDS_TIMEOUT_MS ?? 900_000), }, memory: memoryV2?.getStatus ? await Promise.resolve(memoryV2.getStatus()) : { enabled: Boolean(conversationMemoryService?.isEnabled?.()), backend: 'legacy', selectedBackend: conversationMemoryService ? 'legacy-conversation-memory' : null, profileEnabled: false, eventLogEnabled: Boolean(conversationMemoryService?.isEnabled?.()), vectorEnabled: false, failOpen: true, backends: [ { name: 'legacy-conversation-memory', available: Boolean(conversationMemoryService), supports: { resolve: Boolean(conversationMemoryService?.listMemories), write: Boolean(conversationMemoryService?.saveAndAnalyze), compact: Boolean(conversationMemoryService?.analyzeUser), }, }, ], }, targets: targetStatuses, }; } async function resolveUserMemories( userId, { sessionId = null, query = null, limit = null, forceDeepReasoning = false, recallQuestion = false } = {}, ) { if (!userId) return []; const intervention = resolveMemoryInterventionMode({ forceDeepReasoning, recallQuestion, context: 'agent', }); const resolvedLimit = Number.isFinite(Number(limit)) && Number(limit) > 0 ? Number(limit) : memoryLimitForIntervention(intervention, { context: 'agent' }); return resolveMemoriesWithLegacyFallback({ memoryV2, conversationMemoryService, userId, sessionId, query, limit: resolvedLimit, recallQuestion, }); } async function startSessionForUser( userId, { workingDir, sessionPolicy = null, recipe = null, } = {}, ) { const resolvedWorkingDir = workingDir ?? await userAuth.resolveWorkingDir(userId); const resolvedSessionPolicy = sessionPolicy ?? await userAuth.getAgentSessionPolicy(userId); const startTarget = await pickTarget(); const upstream = await apiFetch(startTarget, apiSecret, '/agent/start', { method: 'POST', body: JSON.stringify({ ...(resolvedWorkingDir ? { working_dir: resolvedWorkingDir } : {}), enable_context_memory: resolvedSessionPolicy?.enableContextMemory, ...(resolvedSessionPolicy?.extensionOverrides ? { extension_overrides: resolvedSessionPolicy.extensionOverrides } : {}), ...(recipe ? { recipe } : {}), }), }); const text = await upstream.text(); if (!upstream.ok) { throw new Error(text || `创建会话失败 (${upstream.status})`); } const session = JSON.parse(text); if (!session?.id) { throw new Error('创建会话失败：缺少 session id'); } await rememberSessionTarget(session.id, startTarget); await sessionStore.registerAgentSession(userId, session.id, startTarget); if (resolvedSessionPolicy?.gooseMode) { const modeRes = await apiFetch(startTarget, apiSecret, '/agent/update_session', { method: 'POST', body: JSON.stringify({ session_id: session.id, goose_mode: resolvedSessionPolicy.gooseMode, }), }); if (!modeRes.ok) { const modeText = await modeRes.text().catch(() => ''); throw new Error(modeText || `设置会话模式失败 (${modeRes.status})`); } } const publishLayout = await userAuth.getUserPublishLayout(userId); const userMemories = await resolveUserMemories(userId, { sessionId: session.id }); await reconcileAgentSession( (pathname, init) => apiFetch(startTarget, apiSecret, pathname, init), session.id, { workingDir: resolvedWorkingDir, sessionPolicy: resolvedSessionPolicy, sandboxConstraints: publishLayout?.constraints ?? null, userMemories, userContext: publishLayout ? { userId, displayName: publishLayout.displayName, username: publishLayout.username, slug: publishLayout.slug, } : null, }, ); await applySessionLlmProvider(session.id); return session; } async function resolveTarget(sessionId) { if (targets.length <= 1 || !sessionId) return primaryTarget; try { const routedTarget = await runtimeRouter?.resolveSessionTarget(sessionId); if (routedTarget) return routedTarget; const { target, node } = await sessionStore.getSessionTarget(sessionId); // Prefer the pinned URL: stable across reordering/resizing the target list. // Only honor it if that upstream is still configured; otherwise fall back to // the legacy integer index, then to primary. if (target && targets.includes(target)) return target; if (brokerMetrics) { if (target && !targets.includes(target)) { brokerMetrics.resolveTargetMiss({ sessionId, target, node, reason: 'stale_pinned_target', }); } } return targets[node] ?? primaryTarget; } catch { return primaryTarget; } } async function applySessionLlmProvider(sessionId) { if (!llmProviderService || !sessionId) return null; try { const target = await resolveTarget(sessionId); return await llmProviderService.applyBestProviderForSession( sessionId, (url, init) => apiFetch(target, apiSecret, `${url.pathname}${url.search}`, init), ); } catch (err) { console.warn( 'LLM provider apply skipped:', err instanceof Error ? err.message : err, ); return null; } } async function applyLocalFallbackForSession(sessionId) { if (!llmProviderService || !sessionId) return null; const target = await resolveTarget(sessionId); return llmProviderService.applyLocalFallbackForSession( sessionId, (url, init) => apiFetch(target, apiSecret, `${url.pathname}${url.search}`, init), ); } // Tracks which sessions are currently using the vision provider so we only // issue a switch-back call when the session was actually on vision. const visionActiveSessions = new Set(); async function applyVisionProviderForSession(sessionId) { if (!llmProviderService || !sessionId) return null; const target = await resolveTarget(sessionId); return llmProviderService.applyVisionProviderForSession( sessionId, (url, init) => apiFetch(target, apiSecret, `${url.pathname}${url.search}`, init), ); } // Two-step vision approach: // Step 1 — Call Qwen VL directly (not through Goose) to analyze the image. // Step 2 — Inject Qwen's text description + server-relative image paths into the // user_message that goes to DeepSeek via Goose. DeepSeek retains full // tool-calling capability (write_file, etc.) and creates the page properly. async function buildVisionBody(userMessage, userId, publishLayout) { return buildVisionPayload({ userMessage, userId, publishLayout, localFetchAsset, llmProviderService, imgproxySigner, }); } async function buildAttachmentBody(userMessage, userId) { return buildAttachmentPayload({ userMessage, userId, localFetchAsset, }); } async function getSessionPolicyForToolMode(userId, toolMode = 'chat') { if (toolMode === 'code' && userAuth.getCodeAgentSessionPolicy) { return userAuth.getCodeAgentSessionPolicy(userId); } return userAuth.getAgentSessionPolicy(userId); } async function reconcileSessionPolicyForUser( userId, sessionId, { toolMode = 'chat', query = null, forceDeepReasoning = false } = {}, ) { if (!userId || !sessionId) return; const target = await resolveTarget(sessionId); const workingDir = await userAuth.resolveWorkingDir(userId); const sessionPolicy = await getSessionPolicyForToolMode(userId, toolMode); const publishLayout = await userAuth.getUserPublishLayout(userId); const userMemories = await resolveUserMemories(userId, { sessionId, query, forceDeepReasoning, }); await reconcileAgentSession( (pathname, init) => apiFetch(target, apiSecret, pathname, init), sessionId, { workingDir, sessionPolicy, sandboxConstraints: publishLayout?.constraints ?? null, tolerateInvalidWorkingDir: true, userMemories, userContext: publishLayout ? { userId, displayName: publishLayout.displayName, username: publishLayout.username, slug: publishLayout.slug, } : null, }, ); } async function syncHistoricalImageTurnIsolation(sessionId, activeMessageId) { const activeId = String(activeMessageId ?? '').trim(); if (!sessionId || !activeId) return { changed: false, updated: false }; try { const target = await resolveTarget(sessionId); const upstream = await apiFetch( target, apiSecret, `/sessions/${encodeURIComponent(sessionId)}`, ); if (!upstream.ok) { return { changed: false, updated: false, status: upstream.status }; } const session = await upstream.json().catch(() => null); const { conversation, changed } = scrubConversationHistoricalImageAttachments( session?.conversation ?? [], activeId, ); if (!changed) return { changed: false, updated: false }; const update = await apiFetch( target, apiSecret, `/sessions/${encodeURIComponent(sessionId)}`, { method: 'PUT', body: JSON.stringify({ conversation }), }, ); if (!update.ok) { console.warn( `Historical image scrub skipped for session ${sessionId}: upstream ${update.status}`, ); return { changed: true, updated: false, status: update.status }; } return { changed: true, updated: true, status: update.status }; } catch (err) { console.warn( 'Historical image scrub skipped:', err instanceof Error ? err.message : err, ); return { changed: false, updated: false, error: err }; } } async function fetchSessionConversationForUser(userId, sessionId) { if (!userId || !sessionId) return []; const owns = await sessionStore.validateOwnership(userId, sessionId); if (!owns) return []; const target = await resolveTarget(sessionId); const upstream = await apiFetch( target, apiSecret, `/sessions/${encodeURIComponent(sessionId)}`, ); if (!upstream.ok) return []; const session = await upstream.json().catch(() => null); return Array.isArray(session?.conversation) ? session.conversation : []; } async function repairSessionToolHistory(sessionId) { if (!sessionId) return { changed: false, updated: false }; const target = await resolveTarget(sessionId); const upstream = await apiFetch( target, apiSecret, `/sessions/${encodeURIComponent(sessionId)}`, ); if (!upstream.ok) { return { changed: false, updated: false, status: upstream.status }; } const session = await upstream.json().catch(() => null); const repair = repairConversationToolHistory(session?.conversation); if (!repair.changed) return { ...repair, updated: false, status: upstream.status }; const update = await apiFetch( target, apiSecret, `/sessions/${encodeURIComponent(sessionId)}`, { method: 'PUT', body: JSON.stringify({ conversation: repair.conversation }), }, ); if (!update.ok) { const freshSessionRequired = update.status === 404 || update.status === 405; const error = new Error( freshSessionRequired ? '当前会话包含不完整的工具调用历史，需要迁移到新会话' : `会话工具历史修复失败 (${update.status})`, ); error.code = freshSessionRequired ? 'SESSION_TOOL_HISTORY_FRESH_SESSION_REQUIRED' : 'SESSION_TOOL_HISTORY_REPAIR_FAILED'; error.retryable = !freshSessionRequired; error.repairedConversation = repair.conversation; throw error; } return { ...repair, updated: true, status: update.status }; } async function prepareSessionReplyBody( userId, sessionId, requestId, userMessage, { toolMode = 'chat', forceDeepReasoning = false, requireHistoricalImageIsolation = false, } = {}, ) { if (!userId || !sessionId) throw new Error('缺少会话信息'); let disclosureDecision = null; try { disclosureDecision = systemDisclosurePolicyService?.evaluate?.({ userMessage, channel: 'goose_proxy', userId, sessionId, }) ?? null; } catch (err) { console.warn( '[TKMindProxy] system disclosure policy evaluation failed open:', err instanceof Error ? err.message : err, ); } if (disclosureDecision?.enforced) { const error = new Error(disclosureDecision.responseText || '不提供本系统内部技术信息'); error.code = 'SYSTEM_TECHNICAL_DISCLOSURE'; error.status = 403; error.retryable = false; error.policyDecision = { policyId: disclosureDecision.policyId, policyVersion: disclosureDecision.policyVersion, reasonCode: disclosureDecision.reasonCode, categories: disclosureDecision.categories, }; throw error; } const owns = await sessionStore.validateOwnership(userId, sessionId); if (!owns) throw new Error('无权访问该会话'); const gate = await userAuth.canUseChat(userId); if (!gate.ok) { const err = new Error(gate.message ?? '余额不足，请充值后继续使用'); err.code = gate.code; err.status = 402; throw err; } await reconcileSessionPolicyForUser(userId, sessionId, { toolMode, query: firstUserText(userMessage), forceDeepReasoning, }); await applySessionLlmProvider(sessionId); await repairSessionToolHistory(sessionId); const user = await userAuth.getUserById(userId); if (!user) throw new Error('用户不存在'); if (requireHistoricalImageIsolation || messageHasImages(userMessage)) { const imageIsolation = await syncHistoricalImageTurnIsolation(sessionId, userMessage?.id); if (requireHistoricalImageIsolation && imageIsolation.changed && !imageIsolation.updated) { const error = new Error( `historical_image_session_update_unsupported:${imageIsolation.status ?? 'unknown'}`, ); error.code = 'HISTORICAL_IMAGE_SESSION_UPDATE_UNSUPPORTED'; throw error; } } let finalUserMessage = userMessage; if (llmProviderService && messageHasImages(userMessage) && await llmProviderService.hasVisionKey()) { const publishLayout = await userAuth.getUserPublishLayout(userId).catch(() => null); const visionResult = await buildVisionBody(userMessage, userId, publishLayout).catch(() => null); if (visionResult?.userMessage) { finalUserMessage = visionResult.userMessage; } if (visionResult?.billableImageCount > 0 && subscriptionService) { await subscriptionService.consumeImageQuota(userId, visionResult.billableImageCount).catch((err) => { console.warn( 'Subscription image quota consume skipped:', err instanceof Error ? err.message : err, ); }); } } if (messageHasFileAttachments(finalUserMessage)) { const attachmentResult = await buildAttachmentBody(finalUserMessage, userId).catch(() => null); if (attachmentResult?.userMessage) { finalUserMessage = attachmentResult.userMessage; } } const policyState = await userAuth.resolveUserPolicies(user); let body = { request_id: requestId, user_message: finalUserMessage, }; if (!policyState.unrestricted) { body = injectTaskRoutingHint( body, await userAuth.getAgentSessionPolicy(userId), ); } body = { ...body, user_message: ensureGooseUserMessageMetadata(body.user_message), }; const target = await resolveTarget(sessionId); return { body, target }; } async function submitSessionReplyForUser( userId, sessionId, requestId, userMessage, options = {}, ) { const { body, target } = await prepareSessionReplyBody( userId, sessionId, requestId, userMessage, options, ); const upstream = await apiFetch( target, apiSecret, `/sessions/${encodeURIComponent(sessionId)}/reply`, { method: 'POST', body: JSON.stringify(body), }, ); if (!upstream.ok) { const text = await upstream.text().catch(() => ''); throw new Error(text || `发送失败 (${upstream.status})`); } return { ok: true }; } async function fetchSessionBillingCost(sessionId) { const target = await resolveTarget(sessionId); const upstream = await apiFetch( target, apiSecret, `/sessions/${encodeURIComponent(sessionId)}`, ); if (!upstream.ok) return null; return upstream.json(); } async function resolveSessionBillingTokenState(sessionId, tokenStateRaw) { return resolveBillingTokenState(tokenStateRaw, { sessionId, fetchSession: fetchSessionBillingCost, }); } async function submitSessionReplyAndAwaitFinishForUser( userId, sessionId, requestId, userMessage, { toolMode = 'chat', forceDeepReasoning = false, timeoutMs = 15 * 60 * 1000 } = {}, ) { const { body, target } = await prepareSessionReplyBody( userId, sessionId, requestId, userMessage, { toolMode, forceDeepReasoning }, ); const eventsResponse = await apiFetch( target, apiSecret, `/sessions/${encodeURIComponent(sessionId)}/events`, { method: 'GET', headers: { Accept: 'text/event-stream' }, }, ); if (!eventsResponse.ok || !eventsResponse.body) { const text = await eventsResponse.text().catch(() => ''); throw new Error(text || `无法建立 session 事件流 (${eventsResponse.status})`); } const finishPromise = consumeSessionEventsUntilFinish(eventsResponse.body, { requestId, timeoutMs, }); // Prevent unhandled rejection from killing the Portal process when reply // setup fails before we await Finish (e.g. provider Error frames arrive later). const trackedFinish = finishPromise.then( (value) => ({ ok: true, value }), (error) => ({ ok: false, error }), ); try { const replyResponse = await apiFetch( target, apiSecret, `/sessions/${encodeURIComponent(sessionId)}/reply`, { method: 'POST', body: JSON.stringify(body), }, ); if (!replyResponse.ok) { const text = await replyResponse.text().catch(() => ''); eventsResponse.body?.cancel?.().catch?.(() => {}); throw new Error(text || `发送失败 (${replyResponse.status})`); } replyResponse.body?.cancel?.().catch?.(() => {}); const finishResult = await trackedFinish; if (!finishResult.ok) throw finishResult.error; const finish = finishResult.value; const tokenState = await resolveSessionBillingTokenState(sessionId, finish.tokenState); if (tokenState && userAuth.billSessionUsage) { await userAuth .billSessionUsage(userId, sessionId, tokenState, requestId) .catch(() => {}); } return { ok: true, ...finish, tokenState }; } catch (err) { eventsResponse.body?.cancel?.().catch?.(() => {}); await trackedFinish.catch?.(() => {}); throw err; } } const requireUser = async (req, res, next) => { try { const session = req.userSession; if (!session) { res.status(401).json({ message: '未登录' }); return; } const me = await userAuth.getMe(req.userToken); if (!me) { res.status(401).json({ message: '登录已过期' }); return; } req.currentUser = me; next(); } catch (err) { res.status(500).json({ message: err instanceof Error ? err.message : '认证失败' }); } }; const ensureChatAllowed = async (req, res, next) => { const gate = await userAuth.canUseChat(req.currentUser.id); if (!gate.ok) { res.status(402).json({ message: gate.message, code: gate.code, balanceCents: gate.balanceCents, minRechargeCents: gate.minRechargeCents, suggestedTiers: gate.suggestedTiers, }); return; } next(); }; const handlers = { 'POST /agent/start': [ requireUser, ensureChatAllowed, async (req, res) => { try { const workingDir = await userAuth.resolveWorkingDir(req.currentUser.id); const sessionPolicy = await userAuth.getAgentSessionPolicy(req.currentUser.id); const startTarget = await pickTarget(); const upstream = await apiFetch(startTarget, apiSecret, '/agent/start', { method: 'POST', body: JSON.stringify({ working_dir: workingDir, enable_context_memory: sessionPolicy.enableContextMemory, ...(sessionPolicy.extensionOverrides ? { extension_overrides: sessionPolicy.extensionOverrides } : {}), ...(req.body?.recipe ? { recipe: req.body.recipe } : {}), }), }); const text = await upstream.text(); if (!upstream.ok) { res.status(upstream.status).send(text); return; } const session = JSON.parse(text); if (session?.id) { await rememberSessionTarget(session.id, startTarget); await sessionStore.registerAgentSession( req.currentUser.id, session.id, startTarget, ); if (sessionPolicy.gooseMode) { const modeRes = await apiFetch(startTarget, apiSecret, '/agent/update_session', { method: 'POST', body: JSON.stringify({ session_id: session.id, goose_mode: sessionPolicy.gooseMode, }), }); if (!modeRes.ok) { const modeText = await modeRes.text().catch(() => ''); res.status(modeRes.status).send(modeText || '设置会话模式失败'); return; } } const publishLayout = await userAuth.getUserPublishLayout(req.currentUser.id); const userMemories = await resolveUserMemories(req.currentUser.id, { sessionId: session.id, }); const api = (pathname, init) => apiFetch(startTarget, apiSecret, pathname, init); try { await reconcileAgentSession(api, session.id, { workingDir, sessionPolicy, sandboxConstraints: publishLayout?.constraints ?? null, userMemories, userContext: publishLayout ? { userId: req.currentUser.id, displayName: publishLayout.displayName, username: publishLayout.username, slug: publishLayout.slug, } : null, }); } catch (reconcileErr) { res.status(500).json({ message: reconcileErr instanceof Error ? `会话策略同步失败：${reconcileErr.message}` : '会话策略同步失败', }); return; } await applySessionLlmProvider(session.id); } res.status(upstream.status).json(session); } catch (err) { res.status(500).json({ message: err instanceof Error ? err.message : '启动会话失败' }); } }, ], 'POST /agent/resume': [ requireUser, ensureChatAllowed, async (req, res) => { try { const sessionId = req.body?.session_id; if (!sessionId) { res.status(400).json({ message: '缺少 session_id' }); return; } const owns = await sessionStore.validateOwnership(req.currentUser.id, sessionId); if (!owns) { res.status(403).json({ message: '无权访问该会话' }); return; } const resumeTarget = await resolveTarget(sessionId); const upstream = await apiFetch(resumeTarget, apiSecret, '/agent/resume', { method: 'POST', body: JSON.stringify(req.body ?? {}), }); const text = await upstream.text(); if (!upstream.ok) { res.status(upstream.status).send(text); return; } const payload = JSON.parse(text); const skipReconcile = req.body?.skip_reconcile === true; if (!skipReconcile) { const workingDir = await userAuth.resolveWorkingDir(req.currentUser.id); const sessionPolicy = await userAuth.getAgentSessionPolicy(req.currentUser.id); const publishLayout = await userAuth.getUserPublishLayout(req.currentUser.id); const userMemories = await resolveUserMemories(req.currentUser.id, { sessionId, }); try { await reconcileAgentSession( (pathname, init) => apiFetch(resumeTarget, apiSecret, pathname, init), sessionId, { workingDir, sessionPolicy, sandboxConstraints: publishLayout?.constraints ?? null, tolerateInvalidWorkingDir: true, userMemories, userContext: publishLayout ? { userId: req.currentUser.id, displayName: publishLayout.displayName, username: publishLayout.username, slug: publishLayout.slug, } : null, }, ); } catch (reconcileErr) { res.status(500).json({ message: reconcileErr instanceof Error ? `会话恢复后策略同步失败：${reconcileErr.message}` : '会话恢复后策略同步失败', }); return; } } await applySessionLlmProvider(sessionId); res.status(upstream.status).json(payload); } catch (err) { res.status(500).json({ message: sanitizeUserFacingProxyMessage( err instanceof Error ? err.message : '恢复会话失败', '恢复会话失败', ), }); } }, ], 'GET /sessions': [ requireUser, async (req, res) => { try { const offset = Math.max(Number(req.query?.offset ?? 0) || 0, 0); const requestedLimit = Number(req.query?.limit ?? 20) || 20; const limit = Math.min(Math.max(requestedLimit, 1), 100); const query = String(req.query?.query ?? '').trim(); const owned = await userAuth.listOwnedSessionIds(req.currentUser.id); const sessionsById = new Map(); let healthyTargets = 0; let lastFailure = null; for (const target of targets) { try { const upstream = await apiFetch(target, apiSecret, '/sessions', { method: 'GET', signal: AbortSignal.timeout(12_000), }); const text = await upstream.text(); if (!upstream.ok) { lastFailure = text || `upstream ${upstream.status}`; continue; } healthyTargets += 1; const payload = JSON.parse(text); for (const item of payload.sessions ?? []) { if (owned.has(item.id)) sessionsById.set(item.id, item); } } catch (err) { lastFailure = sanitizeUserFacingProxyMessage( err instanceof Error ? err.message : '读取会话失败', '读取会话失败', ); } } const directSessionIds = [...owned].filter(isDirectChatSessionId); const ownedMissingFromGoose = [...owned].filter((sessionId) => !sessionsById.has(sessionId)); const snapshotFallbackIds = [...new Set([...directSessionIds, ...ownedMissingFromGoose])]; if (snapshotFallbackIds.length > 0 && sessionSnapshotService?.isEnabled?.()) { for (const sessionId of snapshotFallbackIds) { if (sessionsById.has(sessionId)) continue; const snapshot = await sessionSnapshotService.get(sessionId).catch(() => null); if (!snapshot?.session) continue; sessionsById.set(sessionId, { ...snapshot.session, message_count: snapshot.meta?.synced_msg_count ?? snapshot.messages?.length ?? 0, updated_at: snapshot.meta?.source_updated_at ?? snapshot.session.updated_at, conversation: undefined, }); } } const stillMissingFromGoose = [...owned].filter((sessionId) => !sessionsById.has(sessionId)); if (stillMissingFromGoose.length > 0 && conversationMemoryService?.loadSessionListSummaries) { const dbSummaries = await conversationMemoryService .loadSessionListSummaries(req.currentUser.id, stillMissingFromGoose) .catch(() => new Map()); for (const sessionId of stillMissingFromGoose) { const summary = dbSummaries.get(sessionId); if (!summary) continue; sessionsById.set(sessionId, { id: sessionId, name: summary.title || sessionId, message_count: Number(summary.messageCount ?? 0), updated_at: summary.updatedAt, user_set_name: false, recipe: null, }); } } if (healthyTargets === 0 && owned.size > 0 && sessionsById.size === 0) { res.status(502).json({ message: lastFailure ?? '后端连接失败，请稍后重试' }); return; } if (healthyTargets < targets.length || healthyTargets === 0) { res.setHeader('X-TKMind-Degraded', '1'); } const sessions = [...sessionsById.values()].sort(sortSessionsByRecent); await enrichSessionHistory(sessions, req.currentUser.id); if (typeof userAuth.getSessionOrigins === 'function' && sessions.length > 0) { try { const origins = await userAuth.getSessionOrigins(sessions.map((s) => s.id)); for (const session of sessions) { session.origin = origins.get(session.id) ?? 'h5'; } } catch { // leave sessions untagged on lookup failure } } const summaries = sessions.map(projectSessionSummary).filter((item) => matchesSessionQuery(item, query)); const paged = summaries.slice(offset, offset + limit); res.json({ sessions: paged, page: { total: summaries.length, offset, limit, has_more: offset + paged.length < summaries.length, query, }, }); } catch (err) { res.status(500).json({ message: sanitizeUserFacingProxyMessage( err instanceof Error ? err.message : '读取会话失败', '读取会话失败', ), }); } }, ], }; const sessionScoped = (build) => [ requireUser, async (req, res, next) => { try { const body = req.body ?? (await readJsonBody(req)); req.body = body; const sessionId = extractSessionId(req, body); if (!sessionId) { res.status(400).json({ message: '缺少 session_id' }); return; } const owns = await sessionStore.validateOwnership(req.currentUser.id, sessionId); if (!owns) { res.status(403).json({ message: '无权访问该会话' }); return; } req.agentSessionId = sessionId; await build(req, res, next); } catch (err) { res.status(500).json({ message: sanitizeUserFacingProxyMessage( err instanceof Error ? err.message : '请求失败', '请求失败', ), }); } }, ]; const proxySessionEvents = async (req, res, sessionId, { onAfterFinish, onEvent } = {}) => { const upstreamAbort = new AbortController(); const streamRequestedAt = Date.now(); let clientClosed = false; const replayEnabled = isSessionStreamReplayEnabled() && sessionStreamStore; const initialLastEventId = parseSessionStreamLastEventId(req.get('last-event-id')); let upstreamLastEventId = replayEnabled ? null : initialLastEventId; const abortUpstream = () => { clientClosed = true; upstreamAbort.abort(); }; req.once('close', abortUpstream); const writeReplayToClient = async (events) => { for (const event of events) { if (clientClosed || res.writableEnded) return; await writeClientChunk(formatSessionStreamSseChunk({ id: event.id, data: event.payload, })); } }; let writeClientChunk = async () => {}; let pendingBalance = null; const waitForDrain = () => new Promise((resolve) => res.once('drain', resolve)); writeClientChunk = async (chunk) => { if (res.writableEnded || clientClosed) return; let needsDrain = !res.write(chunk); if (pendingBalance != null && !res.writableEnded && !clientClosed) { needsDrain = !res.write(appendBalanceEvent(pendingBalance)) || needsDrain; pendingBalance = null; } if (needsDrain && !res.writableEnded && !clientClosed) { await waitForDrain(); } }; try { if (replayEnabled && initialLastEventId) { res.status(200); res.setHeader('Content-Type', 'text/event-stream; charset=utf-8'); res.setHeader('Cache-Control', 'no-cache, no-transform'); res.setHeader('Connection', 'keep-alive'); res.setHeader('X-Accel-Buffering', 'no'); res.flushHeaders?.(); let batch = await sessionStreamStore.listEventsForUser(req.currentUser.id, sessionId, { afterEventId: initialLastEventId, }); if (!batch) { res.status(404).end(); return; } if (batch.cursorMiss) { batch = await sessionStreamStore.listEventsForUser(req.currentUser.id, sessionId); } await writeReplayToClient(batch?.events ?? []); if (shouldSkipUpstreamAfterSessionReplay(batch?.events ?? [], { cursorMiss: batch?.cursorMiss })) { req.off('close', abortUpstream); if (!res.writableEnded) res.end(); return; } upstreamLastEventId = resolveUpstreamResumeEventId( batch?.events ?? [], batch?.cursorEvent ?? null, ); } const pathname = `/sessions/${sessionId}/events`; const sessionTarget = await resolveTarget(sessionId); const upstreamHeaders = { Accept: 'text/event-stream' }; if (upstreamLastEventId) { upstreamHeaders['Last-Event-ID'] = upstreamLastEventId; } const upstream = await apiFetch(sessionTarget, apiSecret, pathname, { method: 'GET', signal: upstreamAbort.signal, headers: upstreamHeaders, }); if (clientClosed) return; if (!upstream.ok || !upstream.body) { const text = await upstream.text().catch(() => ''); if (!res.headersSent) { res.status(upstream.status).send(text); } else if (!res.writableEnded) { res.end(); } return; } if (!res.headersSent) { res.status(upstream.status); res.setHeader('Content-Type', 'text/event-stream; charset=utf-8'); res.setHeader('Cache-Control', 'no-cache, no-transform'); res.setHeader('Connection', 'keep-alive'); res.setHeader('X-Accel-Buffering', 'no'); res.flushHeaders?.(); } const billingTransform = createSseBillingTransform({ onFinish: async (event) => { const billingRequestId = event.request_id ?? event.chat_request_id ?? null; const tokenState = await resolveSessionBillingTokenState(sessionId, event.token_state); const result = await userAuth.billSessionUsage( req.currentUser.id, sessionId, tokenState, billingRequestId, ); if (result.ok && result.costCents > 0 && result.balanceCents != null) { pendingBalance = { balanceCents: result.balanceCents, tokensUsed: result.tokensUsed ?? undefined, lastUsage: { inputTokens: result.deltaInputTokens ?? 0, outputTokens: result.deltaOutputTokens ?? 0, costCents: result.costCents, }, }; } // Fire-and-forget snapshot refresh after conversation finishes. if (typeof onAfterFinish === 'function') { void Promise.resolve().then(() => onAfterFinish(sessionId, req.currentUser.id)).catch(() => {}); } }, }); const source = Readable.fromWeb(upstream.body); let firstChunkSeen = false; const firstTokenProbe = new Transform({ transform(chunk, _encoding, callback) { if (!firstChunkSeen && chunk?.length > 0) { firstChunkSeen = true; markFirstTokenObserved(sessionTarget, Date.now() - streamRequestedAt); } callback(null, chunk); }, }); const linkSanitizer = createSessionEventSanitizer(req.currentUser, { onEvent, normalizeReplayIds: replayEnabled, onPersistFrame: replayEnabled ? ({ id, upstreamEventId, payload }) => { void sessionStreamStore.appendEvent({ userId: req.currentUser.id, sessionId, id, upstreamEventId, payload, }).catch(() => {}); } : undefined, }); const clientSink = new Writable({ write(chunk, _encoding, callback) { writeClientChunk(chunk).then(() => callback(), callback); }, }); // 每 20s 发一个注释行保持连接，防止 nginx/代理因静默超时切断 SSE const keepalive = setInterval(() => { if (!res.writableEnded && !clientClosed) { const ok = res.write(': keepalive\n\n'); if (!ok) source.pause(); } }, 20000); res.on('drain', () => source.resume()); await markStreamStarted(sessionId, sessionTarget); let streamCloseStatus = 'closed'; try { try { await pipeline(source, firstTokenProbe, linkSanitizer, billingTransform, clientSink); } catch (err) { streamCloseStatus = clientClosed || upstreamAbort.signal.aborted ? 'aborted' : 'error'; throw err; } } finally { clearInterval(keepalive); req.off('close', abortUpstream); await markStreamEnded(sessionId, sessionTarget, { status: streamCloseStatus }); if (!res.writableEnded) res.end(); } } catch (err) { req.off('close', abortUpstream); if (clientClosed || upstreamAbort.signal.aborted) { if (!res.writableEnded) res.end(); return; } writeSseProxyErrorAndEnd( res, sanitizeUserFacingProxyMessage( err instanceof Error ? err.message : 'SSE 代理失败', 'SSE 代理失败', ), ); } }; const proxyFallback = async (req, res) => { try { const pathname = req.originalUrl.replace(/^\/api/, '') || '/'; if (isNativeH5ApiPath(pathname)) { res.status(404).json({ message: 'H5 本地接口未找到，请确认服务端已更新并重启', code: 'not_found', }); return; } const isAgentRunPath = (req.method === 'POST' && pathname === '/agent/runs') || (req.method === 'GET' && /^\/agent\/runs\/[^/]+$/.test(pathname)) || (req.method === 'GET' && /^\/agent\/runs\/[^/]+\/events$/.test(pathname)); if (isAgentRunPath) { res.status(503).json({ message: 'Agent Run 接口需在 Portal 本地处理，请确认 server.mjs 已更新并重启后端', code: 'AGENT_RUNS_NATIVE_REQUIRED', }); return; } const policyState = await userAuth.resolveUserPolicies(req.currentUser); const capabilityState = await userAuth.resolveUserCapabilities(req.currentUser); const gate = evaluateProxyRequest(req.method, pathname, policyState.policies, { unrestricted: policyState.unrestricted, }); if (!gate.allowed) { res.status(403).json({ message: gate.reason ?? '该 API 已被策略禁止' }); return; } if ( /^\/agent\/harness_(bootstrap|remember)$/.test(pathname) && !capabilityState.unrestricted && !capabilityState.capabilities.context_memory ) { res.status(403).json({ message: '当前账户未开通项目记忆，无法访问该 API' }); return; } const isReplyPath = pathname.match(/^\/sessions\/[^/]+\/reply$/) && req.method === 'POST'; const sessionMatch = pathname.match(/^\/sessions\/([^/]+)/); if (isReplyPath) { res.status(410).json({ message: '聊天提交入口已统一为 POST /agent/runs', code: 'AGENT_RUNS_REQUIRED', }); return; } let baseBody = req.body; if (isReplyPath && !policyState.unrestricted) { const publishLayout = await userAuth .getUserPublishLayout(req.currentUser.id) .catch(() => null); baseBody = injectCurrentTimeAnchor(req.body, publishLayout?.timezone); baseBody = injectTaskRoutingHint( baseBody, await userAuth.getAgentSessionPolicy(req.currentUser.id), ); } // Vision pre-processing: call Qwen VL directly for image analysis, then inject // the description + image paths into the message text for DeepSeek/Goose. // We do NOT switch the Goose session provider — DeepSeek retains full tool-calling // capability so it can write files and return real links. if (isReplyPath && llmProviderService) { const sessionId = sessionMatch?.[1]; if (sessionId) { const hasImages = messageHasImages(req.body?.user_message); if (hasImages && await llmProviderService.hasVisionKey()) { const publishLayout = await userAuth .getUserPublishLayout(req.currentUser?.id) .catch(() => null); const visionResult = await buildVisionBody( baseBody?.user_message ?? req.body?.user_message, req.currentUser?.id, publishLayout, ).catch(() => null); if (visionResult?.userMessage) { baseBody = { ...(baseBody ?? req.body), user_message: visionResult.userMessage }; } if (visionResult?.billableImageCount > 0 && subscriptionService) { await subscriptionService.consumeImageQuota( req.currentUser.id, visionResult.billableImageCount, ).catch((err) => { console.warn( 'Subscription image quota consume skipped:', err instanceof Error ? err.message : err, ); }); } } } } const body = req.method === 'GET' || req.method === 'HEAD' ? undefined : baseBody ? JSON.stringify(baseBody) : undefined; const fallbackTarget = req.goosedTarget ?? (sessionMatch ? await resolveTarget(sessionMatch[1]) : primaryTarget); const upstream = await apiFetch(fallbackTarget, apiSecret, pathname, { method: req.method, body, headers: { Accept: req.get('accept') ?? '*/*', 'Last-Event-ID': req.get('last-event-id') ?? '', }, }); if (req.method === 'GET' && /^\/sessions\/[^/]+$/.test(pathname) && upstream.ok) { const payload = await upstream.json().catch(() => null); if (payload && Array.isArray(payload.conversation)) { const sanitizedConversation = sanitizeSessionConversationPublicHtmlLinks( payload.conversation, req.currentUser, ); const { conversation, page } = paginateSessionConversation( sanitizedConversation, req.query?.history_before, req.query?.history_limit, ); payload.conversation = conversation; payload.conversation_page = page; } res.status(upstream.status).json(payload ?? {}); return; } sendProxyResponse(res, upstream); } catch (err) { res.status(502).json({ message: sanitizeUserFacingProxyMessage( err instanceof Error ? err.message : '代理失败', '代理失败', ), }); } }; return { requireUser, ensureChatAllowed, applySessionLlmProvider, applyLocalFallbackForSession, applyVisionProviderForSession, reconcileSessionPolicyForUser, handlers, sessionScoped, proxyFallback, proxySessionEvents, resolveTarget, startSessionForUser, fetchSessionConversationForUser, getRuntimeStatus, submitSessionReplyForUser, submitSessionReplyAndAwaitFinishForUser, apiFetch: async (pathname, init) => apiFetch(await pickTarget(), apiSecret, pathname, init), apiFetchTo: (target, pathname, init) => apiFetch(target, apiSecret, pathname, init), }; } export function matchHandler(handlers, method, path) { const key = `${method.toUpperCase()} ${path}`; return handlers[key] ?? null; }