Compare commits
2 Commits
d3239ff292
...
bdeab23b83
| Author | SHA1 | Date | |
|---|---|---|---|
| bdeab23b83 | |||
| 614837e199 |
@@ -2,6 +2,11 @@ import crypto from 'node:crypto';
|
||||
import fs from 'node:fs/promises';
|
||||
import path from 'node:path';
|
||||
import { isDirectChatSessionId } from './direct-chat-service.mjs';
|
||||
import {
|
||||
loadSnapshotMessages,
|
||||
persistSessionTranscriptFromSnapshot,
|
||||
persistSessionTranscriptMessages,
|
||||
} from './conversation-transcript-persist.mjs';
|
||||
|
||||
const DEFAULT_RUN_RETRY_DELAYS_MS = [1500, 5000, 15000];
|
||||
const TERMINAL_STATUSES = new Set(['succeeded', 'failed']);
|
||||
@@ -225,6 +230,7 @@ export function createAgentRunGateway({
|
||||
directChatService = null,
|
||||
chatIntentRouter = null,
|
||||
sessionSnapshotService = null,
|
||||
conversationMemoryService = null,
|
||||
retryDelaysMs = DEFAULT_RUN_RETRY_DELAYS_MS,
|
||||
autoDispatch = envFlag(process.env.MEMIND_AGENT_RUN_AUTODISPATCH, true),
|
||||
maxConcurrentRuns = positiveInteger(
|
||||
@@ -546,7 +552,9 @@ export function createAgentRunGateway({
|
||||
}
|
||||
|
||||
let sessionId = row.agent_session_id ?? null;
|
||||
let escalatedDirectSessionId = null;
|
||||
if (isDirectChatSessionId(sessionId)) {
|
||||
escalatedDirectSessionId = sessionId;
|
||||
await appendEvent(runId, 'direct_session_escalated_to_deep_reasoning', {
|
||||
previousSessionId: sessionId,
|
||||
});
|
||||
@@ -568,8 +576,39 @@ export function createAgentRunGateway({
|
||||
toolMode: runOptions.toolMode,
|
||||
taskType: runOptions.taskType,
|
||||
});
|
||||
if (escalatedDirectSessionId) {
|
||||
const priorMessages = await loadSnapshotMessages(
|
||||
sessionSnapshotService,
|
||||
escalatedDirectSessionId,
|
||||
);
|
||||
const persisted = await persistSessionTranscriptMessages({
|
||||
conversationMemoryService,
|
||||
sessionId,
|
||||
userId: row.user_id,
|
||||
messages: priorMessages,
|
||||
});
|
||||
if (persisted.saved > 0) {
|
||||
await appendEvent(runId, 'direct_session_transcript_persisted', {
|
||||
previousSessionId: escalatedDirectSessionId,
|
||||
sessionId,
|
||||
saved: persisted.saved,
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const transcriptPersisted = await persistSessionTranscriptFromSnapshot({
|
||||
sessionSnapshotService,
|
||||
conversationMemoryService,
|
||||
sessionId,
|
||||
userId: row.user_id,
|
||||
});
|
||||
if (transcriptPersisted.saved > 0) {
|
||||
await appendEvent(runId, 'portal_direct_transcript_persisted', {
|
||||
sessionId,
|
||||
saved: transcriptPersisted.saved,
|
||||
});
|
||||
}
|
||||
await invalidatePortalDirectChatSnapshot(sessionId);
|
||||
await tkmindProxy.submitSessionReplyForUser(
|
||||
row.user_id,
|
||||
|
||||
@@ -640,6 +640,140 @@ test('agent run escalates direct sessions to a new backend session when forced',
|
||||
assert.ok(pool.events.some((event) => event.eventType === 'direct_session_escalated_to_deep_reasoning'));
|
||||
});
|
||||
|
||||
test('agent run persists direct session transcript before escalating to goosed', async () => {
|
||||
const pool = createFakePool();
|
||||
const submitted = [];
|
||||
const saved = [];
|
||||
const removed = [];
|
||||
const gateway = createAgentRunGateway({
|
||||
pool,
|
||||
tkmindProxy: {
|
||||
async startSessionForUser(userId) {
|
||||
assert.equal(userId, 'user-1');
|
||||
return { id: 'deep-session-1' };
|
||||
},
|
||||
async submitSessionReplyForUser(userId, sessionId, requestId, userMessage, options = {}) {
|
||||
submitted.push({ userId, sessionId, requestId, userMessage, options });
|
||||
},
|
||||
},
|
||||
sessionSnapshotService: {
|
||||
async get(sessionId) {
|
||||
if (sessionId !== 'h5direct_existing') return null;
|
||||
return {
|
||||
messages: [
|
||||
{ role: 'user', content: [{ type: 'text', text: '中考政策' }] },
|
||||
{ role: 'assistant', content: [{ type: 'text', text: '政策摘要' }] },
|
||||
],
|
||||
};
|
||||
},
|
||||
async remove(sessionId) {
|
||||
removed.push(sessionId);
|
||||
},
|
||||
},
|
||||
conversationMemoryService: {
|
||||
async saveConversationMessages(sessionId, userId, messages) {
|
||||
saved.push({ sessionId, userId, messages });
|
||||
return messages;
|
||||
},
|
||||
},
|
||||
retryDelaysMs: [],
|
||||
});
|
||||
|
||||
const run = await gateway.createRun('user-1', {
|
||||
sessionId: 'h5direct_existing',
|
||||
requestId: 'req-force-deep-transcript',
|
||||
userMessage: { role: 'user', content: [{ type: 'text', text: '帮我生成页面 public/a.html' }] },
|
||||
forceDeepReasoning: true,
|
||||
});
|
||||
|
||||
await waitFor(() => pool.runs.get(run.id)?.status === 'succeeded');
|
||||
assert.equal(saved.length, 1);
|
||||
assert.equal(saved[0].sessionId, 'deep-session-1');
|
||||
assert.equal(saved[0].messages.length, 2);
|
||||
assert.ok(pool.events.some((event) => event.eventType === 'direct_session_transcript_persisted'));
|
||||
assert.equal(submitted[0].sessionId, 'deep-session-1');
|
||||
assert.equal(removed.length, 1);
|
||||
assert.equal(removed[0], 'deep-session-1');
|
||||
});
|
||||
|
||||
test('agent run persists portal direct snapshot before goosed submit on same session', async () => {
|
||||
const pool = createFakePool();
|
||||
const submitted = [];
|
||||
const saved = [];
|
||||
const gateway = createAgentRunGateway({
|
||||
pool,
|
||||
userAuth: {
|
||||
async getUserCapabilities() {
|
||||
return { grantedSkills: ['static-page-publish'] };
|
||||
},
|
||||
},
|
||||
tkmindProxy: {
|
||||
async submitSessionReplyForUser(userId, sessionId, requestId, userMessage) {
|
||||
submitted.push({ userId, sessionId, requestId, userMessage });
|
||||
},
|
||||
},
|
||||
sessionSnapshotService: {
|
||||
async get(sessionId) {
|
||||
if (sessionId !== '20260705_2') return null;
|
||||
return {
|
||||
messages: [
|
||||
{ role: 'user', content: [{ type: 'text', text: '深度搜索' }] },
|
||||
{ role: 'assistant', content: [{ type: 'text', text: '搜索结果' }], metadata: { source: 'portal-direct-chat' } },
|
||||
],
|
||||
};
|
||||
},
|
||||
async remove(sessionId) {
|
||||
assert.equal(sessionId, '20260705_2');
|
||||
},
|
||||
},
|
||||
conversationMemoryService: {
|
||||
async saveConversationMessages(sessionId, userId, messages) {
|
||||
saved.push({ sessionId, userId, messages });
|
||||
return messages;
|
||||
},
|
||||
},
|
||||
chatIntentRouter: {
|
||||
isEnabled() {
|
||||
return true;
|
||||
},
|
||||
async classify() {
|
||||
return {
|
||||
route: 'agent_orchestration',
|
||||
confidence: 0.93,
|
||||
reason: '需要生成页面',
|
||||
suggestedSkill: 'static-page-publish',
|
||||
source: 'llm',
|
||||
};
|
||||
},
|
||||
applyAgentOrchestration(userMessage, classification) {
|
||||
const displayText = userMessage?.content?.[0]?.text ?? '';
|
||||
return {
|
||||
...userMessage,
|
||||
content: [{
|
||||
type: 'text',
|
||||
text: `【Memind 任务编排】${classification.reason}\n用户任务:${displayText}`,
|
||||
}],
|
||||
};
|
||||
},
|
||||
},
|
||||
retryDelaysMs: [],
|
||||
});
|
||||
|
||||
const run = await gateway.createRun('user-1', {
|
||||
sessionId: '20260705_2',
|
||||
requestId: 'req-portal-direct-persist',
|
||||
userMessage: { role: 'user', content: [{ type: 'text', text: '生成报告' }] },
|
||||
});
|
||||
|
||||
await waitFor(() => pool.runs.get(run.id)?.status === 'succeeded');
|
||||
assert.equal(saved.length, 1);
|
||||
assert.equal(saved[0].sessionId, '20260705_2');
|
||||
assert.equal(saved[0].messages.length, 2);
|
||||
assert.ok(pool.events.some((event) => event.eventType === 'portal_direct_transcript_persisted'));
|
||||
assert.equal(submitted.length, 1);
|
||||
assert.equal(submitted[0].sessionId, '20260705_2');
|
||||
});
|
||||
|
||||
test('agent run with code tool mode starts and submits with code policy', async () => {
|
||||
const pool = createFakePool();
|
||||
const submitted = [];
|
||||
|
||||
@@ -0,0 +1,126 @@
|
||||
/**
|
||||
* Repair Goose conversations from h5_conversation_messages when agent sync
|
||||
* left empty placeholders (portal direct chat → goosed escalation).
|
||||
*/
|
||||
|
||||
import { extractConversationMessageText } from './conversation-memory.mjs';
|
||||
|
||||
export function parseStoredConversationRow(row) {
|
||||
if (!row) return null;
|
||||
const rawJson = row.raw_json ?? row.rawJson;
|
||||
if (typeof rawJson === 'string' && rawJson.trim()) {
|
||||
try {
|
||||
const parsed = JSON.parse(rawJson);
|
||||
if (parsed && typeof parsed === 'object') {
|
||||
return {
|
||||
...parsed,
|
||||
metadata: {
|
||||
userVisible: true,
|
||||
agentVisible: true,
|
||||
...(parsed.metadata && typeof parsed.metadata === 'object' ? parsed.metadata : {}),
|
||||
},
|
||||
};
|
||||
}
|
||||
} catch {
|
||||
// fall through to text reconstruction
|
||||
}
|
||||
}
|
||||
const text = String(row.text ?? '').trim();
|
||||
if (!text) return null;
|
||||
const messageKey = String(row.message_key ?? row.messageKey ?? '').trim();
|
||||
const createdMs = Number(row.created_at ?? row.createdAt ?? 0) || Date.now();
|
||||
return {
|
||||
id: messageKey || undefined,
|
||||
role: String(row.role ?? 'user').trim() || 'user',
|
||||
created: Math.floor(createdMs / 1000),
|
||||
content: [{ type: 'text', text }],
|
||||
metadata: {
|
||||
userVisible: true,
|
||||
agentVisible: true,
|
||||
...(String(row.role ?? '') === 'assistant' ? { source: 'conversation-db-repair' } : {}),
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
export function countNonEmptyConversationMessages(messages) {
|
||||
if (!Array.isArray(messages)) return 0;
|
||||
return messages.filter((message) => {
|
||||
if (message?.metadata?.userVisible === false) return false;
|
||||
return Boolean(extractConversationMessageText(message));
|
||||
}).length;
|
||||
}
|
||||
|
||||
export function buildConversationFromDbRows(dbRows) {
|
||||
if (!Array.isArray(dbRows) || dbRows.length === 0) return [];
|
||||
const byKey = new Map();
|
||||
for (const row of dbRows) {
|
||||
const parsed = parseStoredConversationRow(row);
|
||||
if (!parsed) continue;
|
||||
const key =
|
||||
String(row.message_key ?? row.messageKey ?? parsed.id ?? '').trim() ||
|
||||
`${parsed.role}:${extractConversationMessageText(parsed).slice(0, 48)}`;
|
||||
byKey.set(key, {
|
||||
message: parsed,
|
||||
sequenceNo: Number(row.sequence_no ?? row.sequenceNo ?? 0),
|
||||
createdAt: Number(row.created_at ?? row.createdAt ?? 0),
|
||||
});
|
||||
}
|
||||
return [...byKey.values()]
|
||||
.sort((left, right) => {
|
||||
if (left.sequenceNo !== right.sequenceNo) return left.sequenceNo - right.sequenceNo;
|
||||
if (left.createdAt !== right.createdAt) return left.createdAt - right.createdAt;
|
||||
return String(left.message.id ?? '').localeCompare(String(right.message.id ?? ''));
|
||||
})
|
||||
.map((entry) => entry.message);
|
||||
}
|
||||
|
||||
export function shouldRepairConversationFromDb(gooseMessages, dbRows) {
|
||||
const dbBuilt = buildConversationFromDbRows(dbRows);
|
||||
const dbCount = countNonEmptyConversationMessages(dbBuilt);
|
||||
if (dbCount === 0) return false;
|
||||
|
||||
const gooseVisible = Array.isArray(gooseMessages)
|
||||
? gooseMessages.filter((message) => message?.metadata?.userVisible !== false)
|
||||
: [];
|
||||
const gooseNonEmpty = countNonEmptyConversationMessages(gooseVisible);
|
||||
const gooseEmptyVisible = gooseVisible.some(
|
||||
(message) => !extractConversationMessageText(message),
|
||||
);
|
||||
|
||||
if (gooseNonEmpty === 0) return true;
|
||||
if (dbCount > gooseNonEmpty) return true;
|
||||
if (gooseEmptyVisible && dbCount >= gooseNonEmpty) return true;
|
||||
return false;
|
||||
}
|
||||
|
||||
export function repairConversationFromDbRows(gooseMessages, dbRows) {
|
||||
const dbBuilt = buildConversationFromDbRows(dbRows);
|
||||
if (!shouldRepairConversationFromDb(gooseMessages, dbRows)) {
|
||||
return Array.isArray(gooseMessages) ? gooseMessages : [];
|
||||
}
|
||||
if (dbBuilt.length === 0) return Array.isArray(gooseMessages) ? gooseMessages : [];
|
||||
return dbBuilt;
|
||||
}
|
||||
|
||||
export async function loadSessionConversationRows(pool, sessionId, userId) {
|
||||
if (!pool || !sessionId || !userId) return [];
|
||||
const [rows] = await pool.query(
|
||||
`SELECT message_key, sequence_no, role, text, raw_json, created_at
|
||||
FROM h5_conversation_messages
|
||||
WHERE agent_session_id = ? AND user_id = ?
|
||||
ORDER BY sequence_no ASC, created_at ASC`,
|
||||
[sessionId, userId],
|
||||
);
|
||||
return rows;
|
||||
}
|
||||
|
||||
export async function repairSessionConversationFromDb(pool, gooseSession, sessionId, userId) {
|
||||
if (!gooseSession || !pool || !sessionId || !userId) return gooseSession;
|
||||
const dbRows = await loadSessionConversationRows(pool, sessionId, userId);
|
||||
const conversation = Array.isArray(gooseSession.conversation) ? gooseSession.conversation : [];
|
||||
if (!shouldRepairConversationFromDb(conversation, dbRows)) return gooseSession;
|
||||
return {
|
||||
...gooseSession,
|
||||
conversation: repairConversationFromDbRows(conversation, dbRows),
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,86 @@
|
||||
import test from 'node:test';
|
||||
import assert from 'node:assert/strict';
|
||||
|
||||
import {
|
||||
buildConversationFromDbRows,
|
||||
countNonEmptyConversationMessages,
|
||||
parseStoredConversationRow,
|
||||
repairConversationFromDbRows,
|
||||
shouldRepairConversationFromDb,
|
||||
} from './conversation-repair.mjs';
|
||||
|
||||
const gooseMsg = (id, role, text) => ({
|
||||
id,
|
||||
role,
|
||||
metadata: { userVisible: true },
|
||||
content: text ? [{ type: 'text', text }] : [],
|
||||
});
|
||||
|
||||
test('parseStoredConversationRow prefers raw_json', () => {
|
||||
const row = {
|
||||
message_key: 'abc',
|
||||
role: 'user',
|
||||
text: 'fallback',
|
||||
raw_json: JSON.stringify({
|
||||
id: 'abc',
|
||||
role: 'user',
|
||||
content: [{ type: 'text', text: 'from raw' }],
|
||||
metadata: { userVisible: true, displayText: '可见文案' },
|
||||
}),
|
||||
created_at: 1_700_000_000_000,
|
||||
};
|
||||
const parsed = parseStoredConversationRow(row);
|
||||
assert.equal(parsed.content[0].text, 'from raw');
|
||||
assert.equal(parsed.metadata.displayText, '可见文案');
|
||||
});
|
||||
|
||||
test('shouldRepairConversationFromDb when goose has empty placeholders', () => {
|
||||
const goose = [
|
||||
gooseMsg('u1', 'user', '编排前缀'),
|
||||
gooseMsg('a1', 'assistant', '好的'),
|
||||
gooseMsg('msg_empty1', 'assistant', ''),
|
||||
gooseMsg('msg_empty2', 'user', ''),
|
||||
];
|
||||
const dbRows = [
|
||||
{
|
||||
message_key: 'u-real',
|
||||
role: 'user',
|
||||
text: '小朋友在上海读初中',
|
||||
created_at: 1,
|
||||
sequence_no: 0,
|
||||
},
|
||||
{
|
||||
message_key: 'a-real',
|
||||
role: 'assistant',
|
||||
text: '关于中考选择的分析…',
|
||||
created_at: 2,
|
||||
sequence_no: 1,
|
||||
},
|
||||
];
|
||||
assert.equal(shouldRepairConversationFromDb(goose, dbRows), true);
|
||||
const repaired = repairConversationFromDbRows(goose, dbRows);
|
||||
assert.equal(countNonEmptyConversationMessages(repaired), 2);
|
||||
assert.match(repaired[0].content[0].text, /上海读初中/);
|
||||
assert.match(repaired[1].content[0].text, /中考/);
|
||||
});
|
||||
|
||||
test('shouldRepairConversationFromDb returns false when goose is complete', () => {
|
||||
const goose = [
|
||||
gooseMsg('u1', 'user', 'hello'),
|
||||
gooseMsg('a1', 'assistant', 'hi there'),
|
||||
];
|
||||
const dbRows = [
|
||||
{ message_key: 'u1', role: 'user', text: 'hello', created_at: 1, sequence_no: 0 },
|
||||
];
|
||||
assert.equal(shouldRepairConversationFromDb(goose, dbRows), false);
|
||||
});
|
||||
|
||||
test('buildConversationFromDbRows dedupes by message_key', () => {
|
||||
const rows = [
|
||||
{ message_key: 'same', role: 'user', text: 'first', created_at: 1, sequence_no: 0 },
|
||||
{ message_key: 'same', role: 'user', text: 'second', created_at: 2, sequence_no: 0 },
|
||||
];
|
||||
const built = buildConversationFromDbRows(rows);
|
||||
assert.equal(built.length, 1);
|
||||
assert.equal(built[0].content[0].text, 'second');
|
||||
});
|
||||
@@ -0,0 +1,60 @@
|
||||
/**
|
||||
* Persist portal direct-chat transcripts before Goosed escalation wipes snapshot cache.
|
||||
*/
|
||||
|
||||
import { extractConversationMessageText } from './conversation-memory.mjs';
|
||||
|
||||
export function filterNonemptyUserVisibleMessages(messages) {
|
||||
if (!Array.isArray(messages)) return [];
|
||||
return messages.filter((message) => {
|
||||
if (message?.metadata?.userVisible === false) return false;
|
||||
return Boolean(extractConversationMessageText(message));
|
||||
});
|
||||
}
|
||||
|
||||
export async function loadSnapshotMessages(sessionSnapshotService, sessionId) {
|
||||
if (!sessionSnapshotService?.get || !sessionId) return [];
|
||||
const snapshot = await sessionSnapshotService.get(sessionId).catch(() => null);
|
||||
return Array.isArray(snapshot?.messages) ? snapshot.messages : [];
|
||||
}
|
||||
|
||||
export async function persistSessionTranscriptMessages({
|
||||
conversationMemoryService,
|
||||
sessionId,
|
||||
userId,
|
||||
messages,
|
||||
} = {}) {
|
||||
if (!conversationMemoryService?.saveConversationMessages || !sessionId || !userId) {
|
||||
return { saved: 0, skipped: true };
|
||||
}
|
||||
const persistable = filterNonemptyUserVisibleMessages(messages);
|
||||
if (!persistable.length) return { saved: 0, skipped: true };
|
||||
const saved = await conversationMemoryService.saveConversationMessages(
|
||||
sessionId,
|
||||
userId,
|
||||
persistable,
|
||||
);
|
||||
return { saved: Array.isArray(saved) ? saved.length : persistable.length, skipped: false };
|
||||
}
|
||||
|
||||
export async function persistSessionTranscriptFromSnapshot({
|
||||
sessionSnapshotService,
|
||||
conversationMemoryService,
|
||||
sessionId,
|
||||
userId,
|
||||
sourceSessionId = null,
|
||||
} = {}) {
|
||||
const sourceIds = [...new Set([sourceSessionId, sessionId].filter(Boolean))];
|
||||
let totalSaved = 0;
|
||||
for (const sourceId of sourceIds) {
|
||||
const messages = await loadSnapshotMessages(sessionSnapshotService, sourceId);
|
||||
const result = await persistSessionTranscriptMessages({
|
||||
conversationMemoryService,
|
||||
sessionId,
|
||||
userId,
|
||||
messages,
|
||||
});
|
||||
totalSaved += result.saved;
|
||||
}
|
||||
return { saved: totalSaved };
|
||||
}
|
||||
@@ -0,0 +1,71 @@
|
||||
import test from 'node:test';
|
||||
import assert from 'node:assert/strict';
|
||||
import {
|
||||
filterNonemptyUserVisibleMessages,
|
||||
persistSessionTranscriptFromSnapshot,
|
||||
persistSessionTranscriptMessages,
|
||||
} from './conversation-transcript-persist.mjs';
|
||||
|
||||
test('filterNonemptyUserVisibleMessages drops empty userVisible placeholders', () => {
|
||||
const filtered = filterNonemptyUserVisibleMessages([
|
||||
{ role: 'user', content: [{ type: 'text', text: 'hello' }], metadata: { userVisible: true } },
|
||||
{ role: 'assistant', id: 'msg_c7f5', content: [{ type: 'text', text: '' }], metadata: { userVisible: true } },
|
||||
{ role: 'assistant', content: [{ type: 'text', text: 'reply' }], metadata: { userVisible: true } },
|
||||
{ role: 'system', content: [{ type: 'text', text: 'hidden' }], metadata: { userVisible: false } },
|
||||
]);
|
||||
assert.equal(filtered.length, 2);
|
||||
assert.equal(filtered[0].content[0].text, 'hello');
|
||||
assert.equal(filtered[1].content[0].text, 'reply');
|
||||
});
|
||||
|
||||
test('persistSessionTranscriptMessages skips empty payloads', async () => {
|
||||
const saved = [];
|
||||
const result = await persistSessionTranscriptMessages({
|
||||
conversationMemoryService: {
|
||||
async saveConversationMessages(sessionId, userId, messages) {
|
||||
saved.push({ sessionId, userId, messages });
|
||||
return messages;
|
||||
},
|
||||
},
|
||||
sessionId: 'session-1',
|
||||
userId: 'user-1',
|
||||
messages: [{ role: 'assistant', content: [{ type: 'text', text: '' }] }],
|
||||
});
|
||||
assert.equal(result.saved, 0);
|
||||
assert.equal(saved.length, 0);
|
||||
});
|
||||
|
||||
test('persistSessionTranscriptFromSnapshot copies prior direct session into new goose session', async () => {
|
||||
const saved = [];
|
||||
const snapshots = new Map([
|
||||
[
|
||||
'h5direct_old',
|
||||
{
|
||||
messages: [
|
||||
{ role: 'user', content: [{ type: 'text', text: '中考政策' }] },
|
||||
{ role: 'assistant', content: [{ type: 'text', text: '这是回复' }], metadata: { source: 'portal-direct-chat' } },
|
||||
],
|
||||
},
|
||||
],
|
||||
]);
|
||||
const result = await persistSessionTranscriptFromSnapshot({
|
||||
sessionSnapshotService: {
|
||||
async get(sessionId) {
|
||||
return snapshots.get(sessionId) ?? null;
|
||||
},
|
||||
},
|
||||
conversationMemoryService: {
|
||||
async saveConversationMessages(sessionId, userId, messages) {
|
||||
saved.push({ sessionId, userId, messages });
|
||||
return messages;
|
||||
},
|
||||
},
|
||||
sessionId: '20260705_2',
|
||||
userId: 'user-john',
|
||||
sourceSessionId: 'h5direct_old',
|
||||
});
|
||||
assert.equal(result.saved, 2);
|
||||
assert.equal(saved.length, 1);
|
||||
assert.equal(saved[0].sessionId, '20260705_2');
|
||||
assert.equal(saved[0].messages.length, 2);
|
||||
});
|
||||
@@ -2,6 +2,7 @@
|
||||
# 反向 SSH 隧道:105 通过 127.0.0.1:19081 访问本机 Portal (:8081)
|
||||
set -euo pipefail
|
||||
|
||||
# 103 上由 john 用户的 LaunchAgent 执行;SSH 目标为 105 的 root(~/.ssh/config Host ssh105-public)
|
||||
HOST="${MEMIND_PORTAL_TUNNEL_HOST:-ssh105-public}"
|
||||
LOCAL_PORT="${MEMIND_PORTAL_TUNNEL_LOCAL_PORT:-8081}"
|
||||
REMOTE_PORT="${MEMIND_PORTAL_TUNNEL_REMOTE_PORT:-19081}"
|
||||
|
||||
@@ -0,0 +1,128 @@
|
||||
#!/usr/bin/env node
|
||||
/**
|
||||
* Simulate GET /api/sessions/:id repair for a session using live 103 data.
|
||||
* Usage: node scripts/verify-session-repair-local.mjs [sessionId] [userId]
|
||||
*/
|
||||
import { createConnection } from 'mysql2/promise';
|
||||
import { dirname, join } from 'node:path';
|
||||
import { fileURLToPath } from 'node:url';
|
||||
import {
|
||||
countNonEmptyConversationMessages,
|
||||
loadSessionConversationRows,
|
||||
repairConversationFromDbRows,
|
||||
shouldRepairConversationFromDb,
|
||||
} from '../conversation-repair.mjs';
|
||||
import { extractConversationMessageText } from '../conversation-memory.mjs';
|
||||
|
||||
const __dirname = dirname(fileURLToPath(import.meta.url));
|
||||
const root = join(__dirname, '..');
|
||||
|
||||
const sessionId = process.argv[2] ?? '20260705_2';
|
||||
const userId = process.argv[3] ?? 'a6fb1e97-2b0f-447b-b138-4561d8e5c53e';
|
||||
|
||||
function previewMessage(message, index) {
|
||||
const text = extractConversationMessageText(message).replace(/\s+/g, ' ').slice(0, 90);
|
||||
return `${index}. [${message.role}] ${text || '(empty)'}`;
|
||||
}
|
||||
|
||||
async function load103Env() {
|
||||
const { execSync } = await import('node:child_process');
|
||||
const envText = execSync('ssh -o BatchMode=yes john@58.38.22.103 cat /Users/john/Project/Memind/.env', {
|
||||
encoding: 'utf8',
|
||||
});
|
||||
const dbLine = envText.match(/^DATABASE_URL=(.+)$/m)?.[1]?.trim().replace(/^["']|["']$/g, '');
|
||||
const secret = envText.match(/^TKMIND_SERVER__SECRET_KEY=(.+)$/m)?.[1]?.trim().replace(/^["']|["']$/g, '');
|
||||
if (!dbLine || !secret) throw new Error('missing DATABASE_URL or TKMIND_SERVER__SECRET_KEY on 103');
|
||||
const url = new URL(dbLine);
|
||||
return {
|
||||
pool: await createConnection({
|
||||
host: url.hostname,
|
||||
port: Number(url.port || 3306),
|
||||
user: decodeURIComponent(url.username),
|
||||
password: decodeURIComponent(url.password),
|
||||
database: url.pathname.replace(/^\//, ''),
|
||||
}),
|
||||
secret,
|
||||
};
|
||||
}
|
||||
|
||||
async function fetchGooseSession(secret, sid) {
|
||||
const { execSync } = await import('node:child_process');
|
||||
const json = execSync(
|
||||
`ssh -o BatchMode=yes john@58.38.22.103 'SECRET=$(grep ^TKMIND_SERVER__SECRET_KEY= /Users/john/Project/Memind/.env | cut -d= -f2- | tr -d "\\""); curl -k -sf "https://127.0.0.1:18006/sessions/${sid}" -H "X-Secret-Key: $SECRET"'`,
|
||||
{ encoding: 'utf8', maxBuffer: 10 * 1024 * 1024 },
|
||||
);
|
||||
return JSON.parse(json);
|
||||
}
|
||||
|
||||
async function fetchProductionApi(sessionId) {
|
||||
const { execSync } = await import('node:child_process');
|
||||
const login = execSync(
|
||||
`curl -sk -c /tmp/verify-session-repair-cj.txt -X POST https://m.tkmind.cn/auth/login -H 'Content-Type: application/json' -d '{"username":"john","password":"981122tj"}'`,
|
||||
{ encoding: 'utf8' },
|
||||
);
|
||||
const auth = JSON.parse(login);
|
||||
if (!auth.authenticated) throw new Error('login failed');
|
||||
const detail = execSync(
|
||||
`curl -sk -b /tmp/verify-session-repair-cj.txt 'https://m.tkmind.cn/api/sessions/${encodeURIComponent(sessionId)}'`,
|
||||
{ encoding: 'utf8' },
|
||||
);
|
||||
return JSON.parse(detail);
|
||||
}
|
||||
|
||||
async function main() {
|
||||
console.log(`\n=== 模拟验证 session ${sessionId} (user ${userId}) ===\n`);
|
||||
|
||||
const { pool, secret } = await load103Env();
|
||||
const gooseSession = await fetchGooseSession(secret, sessionId);
|
||||
const dbRows = await loadSessionConversationRows(pool, sessionId, userId);
|
||||
await pool.end();
|
||||
|
||||
const gooseConversation = (gooseSession.conversation ?? []).filter(
|
||||
(m) => m?.metadata?.userVisible !== false,
|
||||
);
|
||||
const repairedConversation = repairConversationFromDbRows(gooseConversation, dbRows);
|
||||
|
||||
console.log('--- Goose 原始(API 发版前会返回这类数据)---');
|
||||
console.log(`消息数: ${gooseConversation.length}, 非空: ${countNonEmptyConversationMessages(gooseConversation)}`);
|
||||
gooseConversation.forEach((m, i) => console.log(previewMessage(m, i)));
|
||||
|
||||
console.log('\n--- DB h5_conversation_messages ---');
|
||||
console.log(`行数: ${dbRows.length}`);
|
||||
dbRows.forEach((row, i) => {
|
||||
const text = String(row.text ?? '').replace(/\s+/g, ' ').slice(0, 90);
|
||||
console.log(`${i}. [${row.role}] seq=${row.sequence_no} ${text || '(empty)'}`);
|
||||
});
|
||||
|
||||
console.log(`\nshouldRepair: ${shouldRepairConversationFromDb(gooseConversation, dbRows)}`);
|
||||
|
||||
console.log('\n--- 本地 repair 后(发版后 API 应接近此结果)---');
|
||||
console.log(`消息数: ${repairedConversation.length}, 非空: ${countNonEmptyConversationMessages(repairedConversation)}`);
|
||||
repairedConversation.forEach((m, i) => console.log(previewMessage(m, i)));
|
||||
|
||||
console.log('\n--- 当前生产 API(未发版,无 repair)---');
|
||||
try {
|
||||
const prod = await fetchProductionApi(sessionId);
|
||||
const prodConv = (prod.conversation ?? []).filter((m) => m?.metadata?.userVisible !== false);
|
||||
console.log(`name: ${prod.name}`);
|
||||
console.log(`消息数: ${prodConv.length}, 非空: ${countNonEmptyConversationMessages(prodConv)}`);
|
||||
prodConv.forEach((m, i) => console.log(previewMessage(m, i)));
|
||||
} catch (err) {
|
||||
console.log(`生产 API 拉取失败: ${err instanceof Error ? err.message : err}`);
|
||||
}
|
||||
|
||||
console.log('\n=== 结论 ===');
|
||||
const repairedNonEmpty = countNonEmptyConversationMessages(repairedConversation);
|
||||
const gooseNonEmpty = countNonEmptyConversationMessages(gooseConversation);
|
||||
if (repairedNonEmpty > gooseNonEmpty) {
|
||||
console.log(`✔ repair 后非空消息 ${gooseNonEmpty} → ${repairedNonEmpty},发版后 john 应能看到完整历史`);
|
||||
} else {
|
||||
console.log(`✖ repair 未改善非空消息数 (${gooseNonEmpty} → ${repairedNonEmpty})`);
|
||||
process.exitCode = 1;
|
||||
}
|
||||
}
|
||||
|
||||
main().catch((err) => {
|
||||
console.error(err);
|
||||
process.exit(1);
|
||||
});
|
||||
+28
-7
@@ -164,6 +164,8 @@ import { createFeedbackService } from './user-feedback.mjs';
|
||||
import { startScheduleReminderWorker } from './schedule-reminder-worker.mjs';
|
||||
import { createLlmProviderService, RELAY_BOOTSTRAP } from './llm-providers.mjs';
|
||||
import { createDirectChatService, isDirectChatSessionId, isPortalDirectChatSnapshot, sendDirectChatSessionEvents, shouldExpirePortalDirectChatSnapshot } from './direct-chat-service.mjs';
|
||||
import { repairSessionConversationFromDb } from './conversation-repair.mjs';
|
||||
import { filterNonemptyUserVisibleMessages } from './conversation-transcript-persist.mjs';
|
||||
import { createManagedChatIntentRouter } from './chat-intent-router.mjs';
|
||||
import { createSessionSnapshotService } from './session-snapshot.mjs';
|
||||
import { createConversationMemoryService } from './conversation-memory.mjs';
|
||||
@@ -583,6 +585,7 @@ async function bootstrapUserAuth() {
|
||||
directChatService,
|
||||
chatIntentRouter,
|
||||
sessionSnapshotService,
|
||||
conversationMemoryService,
|
||||
autoDispatch: ['1', 'true', 'yes', 'on'].includes(
|
||||
String(process.env.MEMIND_AGENT_RUN_AUTODISPATCH ?? '1').trim().toLowerCase(),
|
||||
),
|
||||
@@ -1888,7 +1891,7 @@ async function ensureUserMemoryCapability(req, res) {
|
||||
return capabilityState;
|
||||
}
|
||||
|
||||
async function loadUserVisibleConversation(sessionId) {
|
||||
async function loadUserVisibleConversation(sessionId, userId) {
|
||||
const target = await tkmindProxy.resolveTarget(sessionId);
|
||||
const upstream = await tkmindProxy.apiFetchTo(target, `/sessions/${encodeURIComponent(sessionId)}`, {
|
||||
method: 'GET',
|
||||
@@ -1897,7 +1900,10 @@ async function loadUserVisibleConversation(sessionId) {
|
||||
const message = await upstream.text().catch(() => '');
|
||||
throw new Error(message || '读取会话失败');
|
||||
}
|
||||
const session = await upstream.json();
|
||||
let session = await upstream.json();
|
||||
if (authPool && userId) {
|
||||
session = await repairSessionConversationFromDb(authPool, session, sessionId, userId);
|
||||
}
|
||||
return (session?.conversation ?? []).filter((message) => message?.metadata?.userVisible);
|
||||
}
|
||||
|
||||
@@ -1928,7 +1934,7 @@ api.post('/user-memory/v1/remember-recent', async (req, res) => {
|
||||
}
|
||||
|
||||
try {
|
||||
const messages = await loadUserVisibleConversation(sessionId);
|
||||
const messages = await loadUserVisibleConversation(sessionId, req.currentUser.id);
|
||||
const result = await memoryV2.write({
|
||||
userId: req.currentUser.id,
|
||||
sessionId,
|
||||
@@ -4469,11 +4475,19 @@ api.get('/sessions/:sessionId', async (req, res, next) => {
|
||||
req.currentUser,
|
||||
);
|
||||
// Cache hit — reconstruct a Goose-compatible session response.
|
||||
const cachedGooseSession = {
|
||||
let cachedGooseSession = {
|
||||
...snapshot.session,
|
||||
// Embed only userVisible messages so getSession callers still work.
|
||||
conversation: sanitizedMessages,
|
||||
};
|
||||
if (authPool) {
|
||||
cachedGooseSession = await repairSessionConversationFromDb(
|
||||
authPool,
|
||||
cachedGooseSession,
|
||||
sessionId,
|
||||
req.currentUser.id,
|
||||
);
|
||||
}
|
||||
return res.json(cachedGooseSession);
|
||||
}
|
||||
}
|
||||
@@ -4494,17 +4508,24 @@ api.get('/sessions/:sessionId', async (req, res, next) => {
|
||||
const text = await upstream.text().catch(() => '');
|
||||
return res.status(upstream.status).send(text);
|
||||
}
|
||||
const gooseSession = await upstream.json();
|
||||
let gooseSession = await upstream.json();
|
||||
if (Array.isArray(gooseSession.conversation)) {
|
||||
gooseSession.conversation = sanitizeSessionConversationPublicHtmlLinks(
|
||||
gooseSession.conversation,
|
||||
req.currentUser,
|
||||
);
|
||||
}
|
||||
if (authPool) {
|
||||
gooseSession = await repairSessionConversationFromDb(
|
||||
authPool,
|
||||
gooseSession,
|
||||
sessionId,
|
||||
req.currentUser.id,
|
||||
);
|
||||
}
|
||||
// Write-through: persist snapshot async, don't block the response.
|
||||
if (sessionSnapshotService?.isEnabled()) {
|
||||
const messages = (gooseSession.conversation ?? [])
|
||||
.filter((m) => m.metadata?.userVisible);
|
||||
const messages = filterNonemptyUserVisibleMessages(gooseSession.conversation ?? []);
|
||||
if (messages.length > 0) {
|
||||
void sessionSnapshotService
|
||||
.save(sessionId, req.currentUser.id, gooseSession, messages)
|
||||
|
||||
@@ -67,6 +67,8 @@ function resolveDisplayTitle(session, messages) {
|
||||
return deriveTitleFromMessages(messages);
|
||||
}
|
||||
|
||||
import { filterNonemptyUserVisibleMessages } from './conversation-transcript-persist.mjs';
|
||||
|
||||
async function syncConversationMemory({
|
||||
memoryV2 = null,
|
||||
conversationMemoryService = null,
|
||||
@@ -279,9 +281,8 @@ export function createSessionSnapshotService(pool, options = {}) {
|
||||
});
|
||||
if (!res.ok) return;
|
||||
const gooseSession = await res.json();
|
||||
// Store raw userVisible messages — same as write-through path.
|
||||
const messages = (gooseSession.conversation ?? [])
|
||||
.filter((m) => m.metadata?.userVisible);
|
||||
// Store userVisible messages with content — skip empty Goose placeholders.
|
||||
const messages = filterNonemptyUserVisibleMessages(gooseSession.conversation ?? []);
|
||||
await save(sessionId, userId, gooseSession, messages);
|
||||
} catch (err) {
|
||||
console.warn('[snapshot] refresh failed:', sessionId, err instanceof Error ? err.message : err);
|
||||
|
||||
@@ -1125,7 +1125,6 @@ export function useTKMindChat(
|
||||
setMessages([]);
|
||||
setChatState('idle');
|
||||
} else if (isTransientConnectError(err)) {
|
||||
clearStoredSessionId(userRef.current?.id);
|
||||
setSession(null);
|
||||
messageHistoryLoadedCountRef.current = 0;
|
||||
messageHistoryTotalRef.current = 0;
|
||||
@@ -1134,7 +1133,7 @@ export function useTKMindChat(
|
||||
setMessageHistoryTotal(0);
|
||||
setMessages([]);
|
||||
setChatState('idle');
|
||||
setNotice('上次会话恢复超时,已为你准备新对话,可直接发送消息');
|
||||
setNotice('上次会话恢复超时,请点左上角 ☰ 选择历史对话,或稍后刷新重试');
|
||||
} else {
|
||||
throw err;
|
||||
}
|
||||
|
||||
+4
-2
@@ -1611,8 +1611,10 @@ export function createTkmindProxy({
|
||||
}
|
||||
|
||||
const directSessionIds = [...owned].filter(isDirectChatSessionId);
|
||||
if (directSessionIds.length > 0 && sessionSnapshotService?.isEnabled?.()) {
|
||||
for (const sessionId of directSessionIds) {
|
||||
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;
|
||||
|
||||
Reference in New Issue
Block a user