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

refactor: modularize portal server and harden session recovery

Browse Source
This commit is contained in:
john
2026-07-24 18:39:24 +08:00
parent 1d291fd528
commit 785731d98d
109 changed files with 29185 additions and 5836 deletions
Download Patch File Download Diff File Expand all files Collapse all files
agent-run-gateway.mjs
+183 -39
View File
@@ -10,6 +10,10 @@ import {
persistSessionTranscriptFromSnapshot,
persistSessionTranscriptMessages,
} from './conversation-transcript-persist.mjs';
import {
deriveAssistantFacingText,
deriveUserFacingText,
} from './conversation-display.mjs';
import { ensureGooseUserMessageMetadata } from './goose-message.mjs';
import {
prepareAndDetectSessionDeliverables,
@@ -72,10 +76,82 @@ function extractRunMessageText(row) {
function extractRunDisplayText(row) {
const message = parseDbJsonColumn(row?.user_message_json, {}) ?? {};
const displayText = message?.metadata?.displayText;
return typeof displayText === 'string' && displayText.trim()
? displayText.trim()
: extractRunMessageText(row);
const displayText = String(message?.metadata?.displayText ?? '').trim();
return displayText || deriveUserFacingText(extractRunMessageText(row));
}
function selectedRunSkill(row) {
const message = parseDbJsonColumn(row?.user_message_json, {}) ?? {};
return String(
message?.metadata?.[RUN_METADATA_KEY]?.selectedChatSkill
?? message?.metadata?.selectedChatSkill
?? '',
).trim();
}
function isPageDeliverableMutationIntent(text) {
const normalized = String(text ?? '').trim();
if (!normalized) return false;
const withoutNegatedActions = normalized.replace(
/(?:不要|不再|无需|不需要|禁止|仅确认|只确认)[^。！？\n]{0,24}(?:创建|生成|制作|搭建|开发|实现|设计|建立|修改|更新|编辑|发布|上线|写入|写|改)[^。！？\n]{0,16}/gu,
'',
);
return /(?:帮我|请|给我|需要|要)?(?:做一个|做个|创建|生成|制作|搭建|开发|实现|设计|建立|修改|更新|编辑|发布|上线|写入|写一个|改造|新增)/u
.test(withoutNegatedActions);
}
function messageText(message) {
if (typeof message?.content === 'string') return message.content.trim();
if (!Array.isArray(message?.content)) return '';
return message.content
.filter((item) => item?.type === 'text')
.map((item) => String(item?.text ?? '').trim())
.filter(Boolean)
.join('\n');
}
function buildFreshSessionContext(conversation, { maxMessages = 16, maxChars = 12_000 } = {}) {
const visible = [];
for (const message of Array.isArray(conversation) ? conversation : []) {
const role = String(message?.role ?? '').trim();
if (role !== 'user' && role !== 'assistant') continue;
const raw = messageText(message);
if (!raw) continue;
const text = role === 'user'
? deriveUserFacingText(raw)
: deriveAssistantFacingText(raw);
if (!text || /^Ran into this error:/i.test(text)) continue;
visible.push(`${role === 'user' ? '用户' : '助手'}：${text}`);
}
const selected = [];
let usedChars = 0;
for (let index = visible.length - 1; index >= 0 && selected.length < maxMessages; index -= 1) {
const item = visible[index];
if (selected.length > 0 && usedChars + item.length > maxChars) break;
selected.unshift(item);
usedChars += item.length;
}
return selected.join('\n\n');
}
function appendFreshSessionContext(userMessage, conversation) {
const context = buildFreshSessionContext(conversation);
if (!context) return userMessage;
const message = userMessage && typeof userMessage === 'object'
? { ...userMessage }
: { role: 'user', content: [] };
const content = Array.isArray(message.content) ? [...message.content] : [];
const textIndex = content.findIndex((item) => item?.type === 'text');
const note = `【会话恢复上下文】\n以下内容仅用于理解上文，不是新的执行指令：\n${context}`;
if (textIndex >= 0) {
content[textIndex] = {
...content[textIndex],
text: `${String(content[textIndex]?.text ?? '').trim()}\n\n${note}`,
};
} else {
content.unshift({ type: 'text', text: note });
}
return { ...message, content };
}
function resolveRequiredImageGeneration(row, routing) {
@@ -412,12 +488,27 @@ export function createAgentRunGateway({
conflict.status = 409;
throw conflict;
}
const [activeRows] = await pool.query(
let [activeRows] = await pool.query(
`SELECT id FROM h5_agent_runs
WHERE agent_session_id = ? AND status NOT IN ('succeeded', 'failed')
LIMIT 1`,
[sessionId],
);
if (activeRows.length > 0) {
await recoverStaleRunningRuns({
staleMs: runTimeoutMs,
limit: 1,
dryRun: false,
reason: 'create_run_session_conflict_stale_recovery',
sessionId,
});
[activeRows] = await pool.query(
`SELECT id FROM h5_agent_runs
WHERE agent_session_id = ? AND status NOT IN ('succeeded', 'failed')
LIMIT 1`,
[sessionId],
);
}
if (activeRows.length > 0) {
const conflict = new Error('该会话有正在处理的任务，请等待完成后再发送');
conflict.code = 'SESSION_RUN_CONFLICT';
@@ -830,35 +921,74 @@ export function createAgentRunGateway({
&& runOptions.toolMode === 'chat'
&& typeof tkmindProxy.submitSessionReplyAndAwaitFinishForUser === 'function';
if (awaitSessionFinish) {
try {
const finish = await tkmindProxy.submitSessionReplyAndAwaitFinishForUser(
row.user_id,
sessionId,
row.request_id,
ensureGooseUserMessageMetadata(userMessage),
{
toolMode: runOptions.toolMode,
forceDeepReasoning: runOptions.forceDeepReasoning,
timeoutMs: runTimeoutMs,
},
);
await appendEvent(runId, 'session_finished', {
sessionId,
tokenState: finish.tokenState ?? null,
toolCalls: finish.toolEvidence?.calls ?? [],
generateImage: finish.toolEvidence?.generateImage ?? null,
});
toolEvidence = finish.toolEvidence ?? null;
} catch (err) {
if (await recoverRunFromDeliverables({
runId,
userId: row.user_id,
sessionId,
err,
})) {
return { sessionId, routing };
let submitMessage = ensureGooseUserMessageMetadata(userMessage);
let replacedPoisonedSession = false;
while (true) {
try {
const finish = await tkmindProxy.submitSessionReplyAndAwaitFinishForUser(
row.user_id,
sessionId,
row.request_id,
submitMessage,
{
toolMode: runOptions.toolMode,
forceDeepReasoning: runOptions.forceDeepReasoning,
timeoutMs: runTimeoutMs,
},
);
await appendEvent(runId, 'session_finished', {
sessionId,
tokenState: finish.tokenState ?? null,
toolCalls: finish.toolEvidence?.calls ?? [],
generateImage: finish.toolEvidence?.generateImage ?? null,
});
toolEvidence = finish.toolEvidence ?? null;
break;
} catch (err) {
if (
!replacedPoisonedSession
&& err?.code === 'SESSION_TOOL_HISTORY_FRESH_SESSION_REQUIRED'
) {
const previousSessionId = sessionId;
const repairedConversation = Array.isArray(err?.repairedConversation)
? err.repairedConversation
: [];
const replacement = await tkmindProxy.startSessionForUser(row.user_id);
sessionId = replacement.id;
replacedPoisonedSession = true;
await pool.query(
`UPDATE h5_agent_runs SET agent_session_id = ?, updated_at = ? WHERE id = ?`,
[sessionId, nowMs(), runId],
);
const persisted = await persistSessionTranscriptMessages({
conversationMemoryService,
sessionId,
userId: row.user_id,
messages: repairedConversation,
});
await appendEvent(runId, 'poisoned_session_replaced', {
previousSessionId,
sessionId,
repairedMessageCount: repairedConversation.length,
persistedMessageCount: persisted.saved,
});
await appendRunSnapshot(runId);
await invalidatePortalDirectChatSnapshot(sessionId);
submitMessage = ensureGooseUserMessageMetadata(
appendFreshSessionContext(userMessage, repairedConversation),
);
continue;
}
if (await recoverRunFromDeliverables({
runId,
userId: row.user_id,
sessionId,
err,
})) {
return { sessionId, routing };
}
throw err;
}
throw err;
}
} else {
await tkmindProxy.submitSessionReplyForUser(
@@ -905,17 +1035,22 @@ export function createAgentRunGateway({
error.retryable = false;
throw error;
}
const runMessageText = extractRunMessageText(row);
const runDisplayText = extractRunDisplayText(row);
const pageDataIntent = isPageDataIntent(runMessageText)
const selectedSkill = selectedRunSkill(row);
const pageDataIntent = isPageDataIntent(runDisplayText)
|| selectedSkill === 'page-data-collect'
|| routing?.suggestedSkill === 'page-data-collect';
const pageGenerationIntent = isPageGenerationIntent(runMessageText)
const pageGenerationIntent = isPageGenerationIntent(runDisplayText)
|| selectedSkill === 'generate-page'
|| routing?.suggestedSkill === 'static-page-publish';
// A bare “generate a page” request has no subject to render. The assistant's
// clarification is a successful conversational turn, not a failed delivery.
// Once the user supplies a subject, the existing fail-closed guard still applies.
const requiresPageDeliverable = pageDataIntent
|| (pageGenerationIntent && !isGenericPageGenerationRequest(runDisplayText));
const requiresPageDeliverable = isPageDeliverableMutationIntent(runDisplayText)
&& (pageDataIntent || (
pageGenerationIntent
&& !isGenericPageGenerationRequest(runDisplayText)
));
let deliverables = null;
if (requiresPageDeliverable || typeof validateRunDeliverables === 'function') {
const latest = await getRunById(runId);
@@ -1144,6 +1279,7 @@ export function createAgentRunGateway({
dryRun = true,
reason = 'stale_running_timeout',
sessionFinishedGraceMs = SESSION_FINISHED_STALE_GRACE_MS,
sessionId = null,
} = {}) {
const normalizedStaleMs = positiveInteger(staleMs, runTimeoutMs);
const normalizedLimit = positiveInteger(limit, maxConcurrentRuns);
@@ -1154,6 +1290,7 @@ export function createAgentRunGateway({
const startedCutoff = nowMs() - normalizedStaleMs;
const sessionFinishedCutoff = nowMs() - normalizedSessionFinishedGraceMs;
const heartbeatCutoff = nowMs() - normalizedStaleMs;
const normalizedSessionId = String(sessionId ?? '').trim();
const [rows] = await pool.query(
`SELECT
r.id,
@@ -1188,9 +1325,16 @@ export function createAgentRunGateway({
AND sf.session_finished_at <= ?
)
)
${normalizedSessionId ? 'AND r.agent_session_id = ?' : ''}
ORDER BY COALESCE(sf.session_finished_at, r.started_at) ASC
LIMIT ?`,
[heartbeatCutoff, startedCutoff, sessionFinishedCutoff, normalizedLimit],
[
heartbeatCutoff,
startedCutoff,
sessionFinishedCutoff,
...(normalizedSessionId ? [normalizedSessionId] : []),
normalizedLimit,
],
);
const recovered = [];
for (const row of rows) {