From a29158d5890ee7dd215772f02192b7ee0a4850d4 Mon Sep 17 00:00:00 2001 From: john Date: Sat, 4 Jul 2026 23:15:44 +0800 Subject: [PATCH] fix: unblock agent events after portal direct chat escalation When a session starts with LLM direct chat then escalates to goosed, invalidate the portal-direct-chat snapshot and stop replaying stale SSE so page-generation turns stream and finish correctly. Co-authored-by: Cursor --- agent-run-gateway.mjs | 8 +++ agent-run-gateway.test.mjs | 57 ++++++++++++++++++ direct-chat-service.mjs | 36 ++++++++++- direct-chat-service.test.mjs | 114 +++++++++++++++++++++++++++++++++++ server.mjs | 21 +++++-- 5 files changed, 231 insertions(+), 5 deletions(-) diff --git a/agent-run-gateway.mjs b/agent-run-gateway.mjs index 37103d4..00302b3 100644 --- a/agent-run-gateway.mjs +++ b/agent-run-gateway.mjs @@ -224,6 +224,7 @@ export function createAgentRunGateway({ toolGateway = null, directChatService = null, chatIntentRouter = null, + sessionSnapshotService = null, retryDelaysMs = DEFAULT_RUN_RETRY_DELAYS_MS, autoDispatch = envFlag(process.env.MEMIND_AGENT_RUN_AUTODISPATCH, true), maxConcurrentRuns = positiveInteger( @@ -409,6 +410,11 @@ export function createAgentRunGateway({ return Array.isArray(caps?.grantedSkills) ? caps.grantedSkills : []; } + async function invalidatePortalDirectChatSnapshot(sessionId) { + if (!sessionId || isDirectChatSessionId(sessionId) || !sessionSnapshotService?.remove) return; + await sessionSnapshotService.remove(sessionId).catch(() => {}); + } + async function resolveRunRouting(row, userMessage, runOptions) { if (!chatIntentRouter?.classify) return null; const enabled = chatIntentRouter.isEnabled @@ -497,6 +503,7 @@ export function createAgentRunGateway({ const workingDir = userAuth?.resolveWorkingDir ? await userAuth.resolveWorkingDir(row.user_id) : undefined; + await invalidatePortalDirectChatSnapshot(row.agent_session_id ?? null); await appendEvent(runId, 'tool_gateway_dispatch', { protocol: toolGatewayStatus.protocol ?? 'agent-run-v1', taskType: runOptions.taskType, @@ -563,6 +570,7 @@ export function createAgentRunGateway({ }); } + await invalidatePortalDirectChatSnapshot(sessionId); await tkmindProxy.submitSessionReplyForUser( row.user_id, sessionId, diff --git a/agent-run-gateway.test.mjs b/agent-run-gateway.test.mjs index 3acc62c..4a76469 100644 --- a/agent-run-gateway.test.mjs +++ b/agent-run-gateway.test.mjs @@ -383,6 +383,63 @@ test('agent run uses direct chat on regular agent sessions when llm routes direc assert.ok(pool.events.some((event) => event.eventType === 'direct_chat_completed')); }); +test('agent run invalidates portal direct chat snapshot before submitting to goosed', async () => { + const pool = createFakePool(); + const submitted = []; + const invalidated = []; + 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 }); + }, + }, + chatIntentRouter: { + isEnabled() { + return true; + }, + async classify() { + return { + route: 'agent_orchestration', + confidence: 1, + reason: '用户开启深度推理', + source: 'rule', + }; + }, + applyAgentOrchestration(userMessage) { + return userMessage; + }, + }, + sessionSnapshotService: { + async remove(sessionId) { + invalidated.push(sessionId); + }, + }, + retryDelaysMs: [], + }); + + const run = await gateway.createRun('user-1', { + sessionId: '20260704_31', + requestId: 'req-essay-page', + forceDeepReasoning: true, + userMessage: { + role: 'user', + content: [{ type: 'text', text: '帮我写一个 200 字散文，做成页面' }], + metadata: { displayText: '帮我写一个 200 字散文，做成页面' }, + }, + }); + + await waitFor(() => pool.runs.get(run.id)?.status === 'succeeded'); + assert.deepEqual(invalidated, ['20260704_31']); + assert.equal(submitted.length, 1); + assert.equal(submitted[0].sessionId, '20260704_31'); +}); + test('agent run records direct_chat_skipped when execution is unavailable', async () => { const pool = createFakePool(); const submitted = []; diff --git a/direct-chat-service.mjs b/direct-chat-service.mjs index 3abffbe..0472d7a 100644 --- a/direct-chat-service.mjs +++ b/direct-chat-service.mjs @@ -198,16 +198,50 @@ function evaluateCanHandle({ return { ok: true, reason: null }; } -export function isPortalDirectChatSnapshot(snapshot) { +export function isPortalDirectChatSnapshot(snapshot, { sessionId = null } = {}) { + if (!snapshot) return false; const messages = Array.isArray(snapshot?.messages) ? snapshot.messages : []; const lastAssistant = [...messages].reverse().find((message) => message?.role === 'assistant'); if (lastAssistant?.metadata?.source !== 'portal-direct-chat') return false; + // Mixed agent sessions: a user turn after the portal-direct-chat assistant means + // execution moved to goosed — never replay the stale direct-chat snapshot. + if (sessionId && !isDirectChatSessionId(sessionId)) { + const lastAssistantIndex = messages.indexOf(lastAssistant); + const hasUserTurnAfterDirectChat = messages + .slice(lastAssistantIndex + 1) + .some((message) => message?.role === 'user'); + if (hasUserTurnAfterDirectChat) return false; + } const updatedAtRaw = snapshot?.session?.updated_at ?? snapshot?.meta?.saved_at ?? 0; const updatedAt = Date.parse(String(updatedAtRaw)) || Number(updatedAtRaw) || 0; if (!updatedAt) return true; return Date.now() - updatedAt <= 5 * 60 * 1000; } +export async function shouldExpirePortalDirectChatSnapshot(pool, sessionId, snapshot) { + if (!pool || !sessionId || !snapshot) return false; + if (isDirectChatSessionId(sessionId)) return false; + if (!isPortalDirectChatSnapshot(snapshot, { sessionId })) return false; + const syncedAt = Number(snapshot?.meta?.synced_at ?? 0); + if (!syncedAt) return false; + const [rows] = await pool.query( + `SELECT r.id + FROM h5_agent_runs r + WHERE r.agent_session_id = ? + AND r.status = 'succeeded' + AND r.completed_at > ? + AND NOT EXISTS ( + SELECT 1 + FROM h5_agent_run_events e + WHERE e.run_id = r.id + AND e.event_type = 'direct_chat_completed' + ) + LIMIT 1`, + [sessionId, syncedAt], + ); + return rows.length > 0; +} + function normalizeUsageForBilling(usage, previousState) { if (!usage) return null; const input = Math.max(0, Number(usage.inputTokens ?? 0) || 0); diff --git a/direct-chat-service.test.mjs b/direct-chat-service.test.mjs index 80dbf68..382afa2 100644 --- a/direct-chat-service.test.mjs +++ b/direct-chat-service.test.mjs @@ -3,7 +3,9 @@ import test from 'node:test'; import { createDirectChatService, isDirectChatSessionId, + isPortalDirectChatSnapshot, sendDirectChatSessionEvents, + shouldExpirePortalDirectChatSnapshot, } from './direct-chat-service.mjs'; test('direct chat is enabled by default', () => { @@ -285,3 +287,115 @@ test('direct chat run saves portal reply into an existing agent session', async assert.equal(saved[1].messages.at(-1)?.metadata?.source, 'portal-direct-chat'); assert.match(saved[1].messages.at(-1)?.content?.[0]?.text ?? '', /太湖/); }); + +test('portal direct chat snapshot replay stays enabled for dedicated h5direct sessions', () => { + const now = new Date().toISOString(); + const snapshot = { + session: { updated_at: now }, + messages: [ + { role: 'user', content: [{ type: 'text', text: 'hi' }] }, + { + role: 'assistant', + metadata: { source: 'portal-direct-chat' }, + content: [{ type: 'text', text: '你好' }], + }, + ], + }; + assert.equal( + isPortalDirectChatSnapshot(snapshot, { sessionId: 'h5direct_abc' }), + true, + ); +}); + +test('portal direct chat snapshot is disabled after agent escalation on regular sessions', () => { + const now = new Date().toISOString(); + const snapshot = { + session: { updated_at: now }, + messages: [ + { role: 'user', content: [{ type: 'text', text: 'hi' }] }, + { + role: 'assistant', + metadata: { source: 'portal-direct-chat' }, + content: [{ type: 'text', text: '你好' }], + }, + { + role: 'user', + content: [{ type: 'text', text: '帮我写一个 200 字散文，做成页面' }], + }, + ], + }; + assert.equal( + isPortalDirectChatSnapshot(snapshot, { sessionId: '20260704_31' }), + false, + ); +}); + +test('portal direct chat snapshot still replays a completed direct turn on regular sessions', () => { + const now = new Date().toISOString(); + const snapshot = { + session: { updated_at: now }, + messages: [ + { role: 'user', content: [{ type: 'text', text: '我想对太湖有更多的了解' }] }, + { + role: 'assistant', + metadata: { source: 'portal-direct-chat' }, + content: [{ type: 'text', text: '太湖是中国第三大淡水湖。' }], + }, + ], + }; + assert.equal( + isPortalDirectChatSnapshot(snapshot, { sessionId: '20260704_11' }), + true, + ); +}); + +test('shouldExpirePortalDirectChatSnapshot detects agent runs after a portal direct turn', async () => { + const now = new Date().toISOString(); + const snapshot = { + session: { updated_at: now }, + meta: { synced_at: 1000 }, + messages: [ + { role: 'user', content: [{ type: 'text', text: 'hi' }] }, + { + role: 'assistant', + metadata: { source: 'portal-direct-chat' }, + content: [{ type: 'text', text: '你好' }], + }, + ], + }; + const pool = { + async query(_sql, params) { + assert.deepEqual(params, ['20260704_31', 1000]); + return [[{ id: 'agent-run-1' }]]; + }, + }; + assert.equal( + await shouldExpirePortalDirectChatSnapshot(pool, '20260704_31', snapshot), + true, + ); +}); + +test('shouldExpirePortalDirectChatSnapshot keeps pure direct chat sessions fast', async () => { + const now = new Date().toISOString(); + const snapshot = { + session: { updated_at: now }, + meta: { synced_at: 1000 }, + messages: [ + { role: 'user', content: [{ type: 'text', text: '太湖' }] }, + { + role: 'assistant', + metadata: { source: 'portal-direct-chat' }, + content: [{ type: 'text', text: '太湖是中国第三大淡水湖。' }], + }, + ], + }; + const pool = { + async query() { + return [[]]; + }, + }; + assert.equal( + await shouldExpirePortalDirectChatSnapshot(pool, '20260704_11', snapshot), + false, + ); +}); diff --git a/server.mjs b/server.mjs index 38e4186..8ab61d4 100644 --- a/server.mjs +++ b/server.mjs @@ -164,7 +164,7 @@ import { createScheduleService } from './schedule-service.mjs'; 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 } from './direct-chat-service.mjs'; +import { createDirectChatService, isDirectChatSessionId, isPortalDirectChatSnapshot, sendDirectChatSessionEvents, shouldExpirePortalDirectChatSnapshot } from './direct-chat-service.mjs'; import { createManagedChatIntentRouter } from './chat-intent-router.mjs'; import { createSessionSnapshotService } from './session-snapshot.mjs'; import { createConversationMemoryService } from './conversation-memory.mjs'; @@ -580,6 +580,7 @@ async function bootstrapUserAuth() { toolGateway, directChatService, chatIntentRouter, + sessionSnapshotService, autoDispatch: ['1', 'true', 'yes', 'on'].includes( String(process.env.MEMIND_AGENT_RUN_AUTODISPATCH ?? '1').trim().toLowerCase(), ), @@ -4470,7 +4471,13 @@ api.get('/sessions/:sessionId', async (req, res, next) => { try { if (sessionSnapshotService?.isEnabled()) { - const snapshot = await sessionSnapshotService.get(sessionId); + let snapshot = await sessionSnapshotService.get(sessionId); + if (snapshot) { + if (authPool && await shouldExpirePortalDirectChatSnapshot(authPool, sessionId, snapshot)) { + await sessionSnapshotService.remove(sessionId).catch(() => {}); + snapshot = null; + } + } if (snapshot) { // REGRESSION GUARD: without both hints, stale snapshot can wipe mid-turn chat. const canUseSnapshotCache = hintMc != null && hintUa != null; @@ -4478,7 +4485,7 @@ api.get('/sessions/:sessionId', async (req, res, next) => { const uaMatch = snapshot.meta.source_updated_at === hintUa; if ( isDirectChatSessionId(sessionId) || - isPortalDirectChatSnapshot(snapshot) || + isPortalDirectChatSnapshot(snapshot, { sessionId }) || (canUseSnapshotCache && mcMatch && uaMatch) ) { const sanitizedMessages = sanitizeSessionConversationPublicHtmlLinks( @@ -4581,7 +4588,13 @@ api.get('/sessions/:sessionId/events', async (req, res, next) => { return sendDirectChatSessionEvents(req, res, snapshot); } const portalDirectSnapshot = await sessionSnapshotService?.get(sessionId).catch(() => null); - if (isPortalDirectChatSnapshot(portalDirectSnapshot)) { + if ( + portalDirectSnapshot && + authPool && + await shouldExpirePortalDirectChatSnapshot(authPool, sessionId, portalDirectSnapshot) + ) { + await sessionSnapshotService?.remove(sessionId).catch(() => {}); + } else if (portalDirectSnapshot && isPortalDirectChatSnapshot(portalDirectSnapshot, { sessionId })) { return sendDirectChatSessionEvents(req, res, portalDirectSnapshot); } const publishDir = resolveMindSpaceUserPublishDir(__dirname, { id: req.currentUser.id });