tkmind/memind
1
0
Fork 0
Code Issues Pull Requests 2 Actions Packages Projects Releases Wiki Activity

feat(h5): web 联网能力、实时查询路由与 session Finish 对齐

Browse Source 
- 新增 web 能力并挂载 platform/web（web_search/fetch_url）
- 实时查询强制 web skill，router fallback 与 await session Finish
- Session Broker 覆盖率/指标、stream replay 与相关单测/E2E 脚本

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
john
2026-07-06 16:06:26 +08:00
parent 08feae8bef
commit 14a00774d9
41 changed files with 3501 additions and 126 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tkmind-proxy.mjs
+218 -49
View File
@@ -1,4 +1,5 @@
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';
@@ -16,13 +17,23 @@ import {
import { buildCurrentTimeAgentPrefix, buildTaskRoutingAgentText } from './user-memory-profile.mjs';
import { reconcileAgentSession } from './session-reconcile.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 { consumeSessionEventsUntilFinish } from './session-reply-wait.mjs';
import {
memoryLimitForIntervention,
resolveMemoryInterventionMode,
} from './memory-intervention.mjs';
import {
formatSessionStreamSseChunk,
isSessionStreamReplayEnabled,
parseSessionSseBlock,
parseSessionStreamLastEventId,
shouldPersistSessionStreamEvent,
shouldSkipUpstreamAfterSessionReplay,
} from './session-stream.mjs';
const insecureDispatcher = new Agent({
connect: { rejectUnauthorized: false },
@@ -566,41 +577,57 @@ export function sanitizeSessionConversationPublicHtmlLinks(conversation, current
return conversation.map((message) => sanitizeSessionMessagePublicHtmlLinks(message, currentUser));
}
function createSessionEventSanitizer(currentUser, { onEvent } = {}) {
function createSessionEventSanitizer(currentUser, { onEvent, onPersistFrame, normalizeReplayIds = false } = {}) {
let buffer = '';
const flushChunk = (controller, chunk) => {
if (!chunk) return;
const block = String(chunk);
if (!block.includes('data: ')) {
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;
const raw = line.slice(6);
let event;
try {
event = JSON.parse(raw);
} catch {
return line;
}
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);
}
return `data: ${JSON.stringify(event)}`;
});
controller.push(sanitizedLines.join('\n'));
controller.push(`${sanitizedLines.join('\n')}\n`);
};
return new Transform({
@@ -844,6 +871,7 @@ export function createTkmindProxy({
apiSecret,
userAuth,
sessionAccess = null,
sessionStreamStore = null,
llmProviderService,
localFetchAsset,
subscriptionService,
@@ -853,6 +881,9 @@ export function createTkmindProxy({
}) {
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;
@@ -1215,6 +1246,16 @@ export function createTkmindProxy({
// 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;
@@ -1319,7 +1360,7 @@ export function createTkmindProxy({
);
}
async function submitSessionReplyForUser(
async function prepareSessionReplyBody(
userId,
sessionId,
requestId,
@@ -1377,8 +1418,24 @@ export function createTkmindProxy({
...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,
@@ -1395,6 +1452,56 @@ export function createTkmindProxy({
return { ok: true };
}
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,
});
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(() => '');
throw new Error(text || `发送失败 (${replyResponse.status})`);
}
replyResponse.body?.cancel?.().catch?.(() => {});
const finish = await finishPromise;
return { ok: true, ...finish };
}
const requireUser = async (req, res, next) => {
try {
const session = req.userSession;
@@ -1745,12 +1852,66 @@ export function createTkmindProxy({
const upstreamAbort = new AbortController();
const streamRequestedAt = Date.now();
let clientClosed = false;
const replayEnabled = isSessionStreamReplayEnabled() && sessionStreamStore;
const initialLastEventId = parseSessionStreamLastEventId(req.get('last-event-id'));
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;
}
}
const pathname = `/sessions/${sessionId}/events`;
const sessionTarget = await resolveTarget(sessionId);
const upstream = await apiFetch(sessionTarget, apiSecret, pathname, {
@@ -1765,18 +1926,23 @@ export function createTkmindProxy({
if (!upstream.ok || !upstream.body) {
const text = await upstream.text().catch(() => '');
res.status(upstream.status).send(text);
if (!res.headersSent) {
res.status(upstream.status).send(text);
} else if (!res.writableEnded) {
res.end();
}
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?.();
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?.();
}
let pendingBalance = null;
const billingTransform = createSseBillingTransform({
onFinish: async (event) => {
const billingRequestId = event.request_id ?? event.chat_request_id ?? null;
@@ -1815,19 +1981,21 @@ export function createTkmindProxy({
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 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);
@@ -2032,6 +2200,7 @@ export function createTkmindProxy({
startSessionForUser,
getRuntimeStatus,
submitSessionReplyForUser,
submitSessionReplyAndAwaitFinishForUser,
apiFetch: async (pathname, init) => apiFetch(await pickTarget(), apiSecret, pathname, init),
apiFetchTo: (target, pathname, init) => apiFetch(target, apiSecret, pathname, init),
};