john
1951 lines
71 KiB
JavaScript
1951 lines
71 KiB
JavaScript
import fs from 'node:fs';
|
||
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 { appendBalanceEvent, createSseBillingTransform } from './sse-billing.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 { createImgproxySigner } from './imgproxy-signer.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 buildPublicStandardImageUrl(rawUrl, userId) {
|
||
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(resolvePublicBaseUrl(), 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) {
|
||
if (res.writableEnded) return;
|
||
if (res.headersSent) {
|
||
try {
|
||
res.write(`event: error\ndata: ${JSON.stringify({ message })}\n\n`);
|
||
} catch {
|
||
// The client may already be gone. Ending below is still harmless.
|
||
}
|
||
res.end();
|
||
return;
|
||
}
|
||
res.status(502).json({ message });
|
||
}
|
||
|
||
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 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 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 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;
|
||
const root = path.resolve(process.cwd(), PUBLISH_ROOT_DIR, normalizedOwner);
|
||
const target = path.resolve(root, normalizedRelativePath);
|
||
if (target !== root && !target.startsWith(`${root}${path.sep}`)) return false;
|
||
return fs.existsSync(target) && fs.statSync(target).isFile();
|
||
}
|
||
|
||
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: canonicalizeStaticPageUrl(publicUrl, relativePath, canonicalRelativePath),
|
||
};
|
||
}
|
||
|
||
export function sanitizePublicHtmlLinksInText(text, currentUser) {
|
||
let next = String(text ?? '').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;
|
||
},
|
||
);
|
||
return next.replace(PUBLIC_HTML_LINK_PATTERN, (match, owner, rawRelativePath) => {
|
||
const result = sanitizeOwnPublicHtmlUrl(match, owner, rawRelativePath, currentUser);
|
||
return result.ok ? result.url : result.notice;
|
||
});
|
||
}
|
||
|
||
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 } = {}) {
|
||
let buffer = '';
|
||
const flushChunk = (controller, chunk) => {
|
||
if (!chunk) return;
|
||
const block = String(chunk);
|
||
if (!block.includes('data: ')) {
|
||
controller.push(block);
|
||
return;
|
||
}
|
||
const lines = block.split('\n');
|
||
const sanitizedLines = lines.map((line) => {
|
||
if (!line.startsWith('data: ')) return line;
|
||
const raw = line.slice(6);
|
||
let event;
|
||
try {
|
||
event = JSON.parse(raw);
|
||
} catch {
|
||
return line;
|
||
}
|
||
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);
|
||
}
|
||
return `data: ${JSON.stringify(event)}`;
|
||
});
|
||
controller.push(sanitizedLines.join('\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) {
|
||
const urls = [];
|
||
const imageUrls = userMessage?.metadata?.imageUrls;
|
||
if (Array.isArray(imageUrls)) {
|
||
urls.push(...imageUrls);
|
||
}
|
||
|
||
const content = userMessage?.content;
|
||
if (Array.isArray(content)) {
|
||
for (const item of content) {
|
||
if (item?.type === 'image_url' && item.image_url?.url) {
|
||
urls.push(item.image_url.url);
|
||
continue;
|
||
}
|
||
if (item?.type !== 'text' || typeof item.text !== 'string') continue;
|
||
for (const line of item.text.split('\n')) {
|
||
const match = line.trim().match(IMAGE_URL_LINE_RE);
|
||
if (match?.[1]) urls.push(match[1].trim());
|
||
}
|
||
}
|
||
}
|
||
|
||
return [...new Set(urls.filter((url) => typeof url === 'string' && url.trim()))];
|
||
}
|
||
|
||
function messageHasImages(userMessage) {
|
||
return extractImageUrlsFromMessage(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,
|
||
}) {
|
||
void imgproxySigner;
|
||
if (!llmProviderService || !userId) return null;
|
||
const rawImageUrls = extractImageUrlsFromMessage(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 undiciFetch(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);
|
||
imageItems.push({
|
||
mimeType,
|
||
data: buffer.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}: <img src="${item.embedUrl}" alt="图片${i + 1}">`)
|
||
.join(' ');
|
||
|
||
const publicUrlPrefix = publishLayout?.publicUrl
|
||
? `${String(publishLayout.publicUrl).replace(/\/$/, '')}/public/`
|
||
: null;
|
||
|
||
const injectedNote =
|
||
'\n\n【TKMind 图片分析结果 — 仅供执行参考，不要向用户复述此段内容】\n' +
|
||
(visionDescription
|
||
? `Qwen VL 图片描述：\n${visionDescription}\n\n`
|
||
: '') +
|
||
'写作约束：不得改写图片里人物的年龄、性别、人数或主体关系；如果用户明确要求儿童语气或童趣风格，也只能调整表达方式，不能把图片主体改写成儿童场景。\n' +
|
||
`图片 HTML 嵌入路径（直接写入 <img> 标签；以下都是无需 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 });
|
||
}
|
||
|
||
return {
|
||
userMessage: {
|
||
...userMessage,
|
||
content: updatedContent,
|
||
metadata: {
|
||
...(userMessage.metadata ?? {}),
|
||
...(originalDisplayText ? { displayText: originalDisplayText } : {}),
|
||
},
|
||
},
|
||
billableImageCount: visionDescription ? 1 : 0,
|
||
};
|
||
}
|
||
|
||
export function createTkmindProxy({
|
||
apiTarget,
|
||
apiTargets,
|
||
apiSecret,
|
||
userAuth,
|
||
llmProviderService,
|
||
localFetchAsset,
|
||
subscriptionService,
|
||
sessionSnapshotService,
|
||
conversationMemoryService,
|
||
memoryV2,
|
||
}) {
|
||
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 "会话 <id>". Best-effort: never blocks the list.
|
||
async function enrichSessionHistory(sessions) {
|
||
if (!sessionSnapshotService?.isEnabled?.()) return;
|
||
const needsSummary = sessions.filter(
|
||
(session) => hasDefaultSessionName(session) || Number(session?.message_count ?? 0) === 0,
|
||
);
|
||
if (needsSummary.length === 0) return;
|
||
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 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 = 40 } = {}) {
|
||
if (!userId) return [];
|
||
if (memoryV2?.resolve) {
|
||
const resolved = await memoryV2.resolve({
|
||
userId,
|
||
sessionId,
|
||
query,
|
||
limit,
|
||
}).catch(() => null);
|
||
return Array.isArray(resolved?.memories) ? resolved.memories : [];
|
||
}
|
||
return conversationMemoryService?.listMemories
|
||
? conversationMemoryService.listMemories(userId, { limit }).catch(() => [])
|
||
: [];
|
||
}
|
||
|
||
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 userAuth.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, limit: 40 });
|
||
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 userAuth.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;
|
||
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 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 } = {},
|
||
) {
|
||
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,
|
||
limit: 40,
|
||
});
|
||
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 submitSessionReplyForUser(
|
||
userId,
|
||
sessionId,
|
||
requestId,
|
||
userMessage,
|
||
{ toolMode = 'chat' } = {},
|
||
) {
|
||
if (!userId || !sessionId) throw new Error('缺少会话信息');
|
||
const owns = await userAuth.ownsSession(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),
|
||
});
|
||
await applySessionLlmProvider(sessionId);
|
||
|
||
const user = await userAuth.getUserById(userId);
|
||
if (!user) throw new 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,
|
||
);
|
||
});
|
||
}
|
||
}
|
||
const policyState = await userAuth.resolveUserPolicies(user);
|
||
let body = {
|
||
request_id: requestId,
|
||
user_message: finalUserMessage,
|
||
};
|
||
if (!policyState.unrestricted) {
|
||
body = injectTaskRoutingHint(
|
||
body,
|
||
await userAuth.getAgentSessionPolicy(userId),
|
||
);
|
||
}
|
||
|
||
const target = await resolveTarget(sessionId);
|
||
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 };
|
||
}
|
||
|
||
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 userAuth.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,
|
||
limit: 40,
|
||
});
|
||
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 userAuth.ownsSession(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,
|
||
limit: 40,
|
||
});
|
||
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 : '读取会话失败',
|
||
'读取会话失败',
|
||
);
|
||
}
|
||
}
|
||
|
||
if (healthyTargets === 0) {
|
||
res.status(502).json({ message: lastFailure ?? '后端连接失败，请稍后重试' });
|
||
return;
|
||
}
|
||
if (healthyTargets < targets.length) {
|
||
res.setHeader('X-TKMind-Degraded', '1');
|
||
}
|
||
|
||
const sessions = [...sessionsById.values()].sort(sortSessionsByRecent);
|
||
await enrichSessionHistory(sessions);
|
||
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 userAuth.ownsSession(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 abortUpstream = () => {
|
||
clientClosed = true;
|
||
upstreamAbort.abort();
|
||
};
|
||
req.once('close', abortUpstream);
|
||
try {
|
||
const pathname = `/sessions/${sessionId}/events`;
|
||
const sessionTarget = await resolveTarget(sessionId);
|
||
const upstream = await apiFetch(sessionTarget, apiSecret, pathname, {
|
||
method: 'GET',
|
||
signal: upstreamAbort.signal,
|
||
headers: {
|
||
Accept: 'text/event-stream',
|
||
'Last-Event-ID': req.get('last-event-id') ?? '',
|
||
},
|
||
});
|
||
if (clientClosed) return;
|
||
|
||
if (!upstream.ok || !upstream.body) {
|
||
const text = await upstream.text().catch(() => '');
|
||
res.status(upstream.status).send(text);
|
||
return;
|
||
}
|
||
|
||
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?.();
|
||
|
||
let pendingBalance = null;
|
||
const billingTransform = createSseBillingTransform({
|
||
onFinish: async (event) => {
|
||
const billingRequestId = event.request_id ?? event.chat_request_id ?? null;
|
||
const result = await userAuth.billSessionUsage(
|
||
req.currentUser.id,
|
||
sessionId,
|
||
event.token_state,
|
||
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 });
|
||
const waitForDrain = () => new Promise((resolve) => res.once('drain', resolve));
|
||
const 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();
|
||
}
|
||
};
|
||
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,
|
||
getRuntimeStatus,
|
||
submitSessionReplyForUser,
|
||
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;
|
||
}
|