diff --git a/agent-run-gateway.mjs b/agent-run-gateway.mjs index 00302b3..656e93e 100644 --- a/agent-run-gateway.mjs +++ b/agent-run-gateway.mjs @@ -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, diff --git a/agent-run-gateway.test.mjs b/agent-run-gateway.test.mjs index 4a76469..fd3c2dc 100644 --- a/agent-run-gateway.test.mjs +++ b/agent-run-gateway.test.mjs @@ -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 = []; diff --git a/conversation-transcript-persist.mjs b/conversation-transcript-persist.mjs new file mode 100644 index 0000000..1ef3c50 --- /dev/null +++ b/conversation-transcript-persist.mjs @@ -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 }; +} diff --git a/conversation-transcript-persist.test.mjs b/conversation-transcript-persist.test.mjs new file mode 100644 index 0000000..7f4f40a --- /dev/null +++ b/conversation-transcript-persist.test.mjs @@ -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); +}); diff --git a/server.mjs b/server.mjs index c650fe5..aaa565d 100644 --- a/server.mjs +++ b/server.mjs @@ -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) diff --git a/session-snapshot.mjs b/session-snapshot.mjs index 114423b..9a4c0b8 100644 --- a/session-snapshot.mjs +++ b/session-snapshot.mjs @@ -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);