fix: persist direct chat transcript before Goosed escalation
Prevent portal direct-chat history from being lost when snapshot cache is invalidated on Goosed submit by writing non-empty messages to DB first, and stop caching empty Goose placeholder messages in snapshots. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
@@ -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,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);
|
||||
});
|
||||
+3
-2
@@ -165,6 +165,7 @@ 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';
|
||||
@@ -584,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(),
|
||||
),
|
||||
@@ -4523,8 +4525,7 @@ api.get('/sessions/:sessionId', async (req, res, next) => {
|
||||
}
|
||||
// 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);
|
||||
|
||||
Reference in New Issue
Block a user