From 1e4f432754f281d5f3dfbb2d4ffd66536653c99b Mon Sep 17 00:00:00 2001 From: john Date: Tue, 7 Jul 2026 11:48:59 +0800 Subject: [PATCH] =?UTF-8?q?fix(chat):=20=E9=98=BB=E6=AD=A2=E5=90=8C?= =?UTF-8?q?=E4=BC=9A=E8=AF=9D=E5=B9=B6=E5=8F=91=20Agent=20Run=20=E5=AF=BC?= =?UTF-8?q?=E8=87=B4=E6=B6=88=E6=81=AF=E9=94=99=E4=B9=B1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 前端 submit 改用 chatStateRef 同步拦截连发；后端 createRun 检测同 session 活跃任务并返回 409，避免多条回复交错落库。 Co-authored-by: Cursor --- agent-run-gateway.mjs | 17 +++++++ agent-run-gateway.test.mjs | 92 ++++++++++++++++++++++++++++++++++++++ agent-run-routes.mjs | 7 +++ agent-run-routes.test.mjs | 37 +++++++++++++++ src/hooks/useTKMindChat.ts | 14 +++--- 5 files changed, 162 insertions(+), 5 deletions(-) diff --git a/agent-run-gateway.mjs b/agent-run-gateway.mjs index 2d0d246..c1f6252 100644 --- a/agent-run-gateway.mjs +++ b/agent-run-gateway.mjs @@ -340,6 +340,23 @@ export function createAgentRunGateway({ return projectRun(existing); } + // Prevent multiple concurrent agent runs on the same Goose session, which would + // cause replies to arrive out of order and appear garbled in the chat UI. + if (sessionId && !isDirectChatSessionId(sessionId)) { + const [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'; + conflict.status = 409; + throw conflict; + } + } + const runId = crypto.randomUUID(); const createdAt = nowMs(); const runMessage = withRunMetadata(userMessage, { diff --git a/agent-run-gateway.test.mjs b/agent-run-gateway.test.mjs index a99df86..fdd6738 100644 --- a/agent-run-gateway.test.mjs +++ b/agent-run-gateway.test.mjs @@ -1,4 +1,5 @@ import assert from 'node:assert/strict'; +import crypto from 'node:crypto'; import fs from 'node:fs/promises'; import os from 'node:os'; import path from 'node:path'; @@ -29,6 +30,13 @@ function createFakePool() { const [userId, requestId] = params; return [[...runs.values()].filter((row) => row.user_id === userId && row.request_id === requestId)]; } + if (sql.includes('agent_session_id = ?') && sql.includes("status NOT IN ('succeeded', 'failed')")) { + const [sessionId] = params; + const active = [...runs.values()].filter( + (row) => row.agent_session_id === sessionId && !['succeeded', 'failed'].includes(row.status), + ); + return [active.slice(0, 1).map((row) => ({ id: row.id }))]; + } if (sql.includes('SELECT * FROM h5_agent_runs WHERE id = ? LIMIT 1')) { return [[runs.get(params[0])].filter(Boolean)]; } @@ -1571,3 +1579,87 @@ test('markRun appends run_snapshot when MEMIND_RUN_STREAM_REPLAY=1', async () => else process.env.MEMIND_RUN_STREAM_REPLAY = previous; } }); + +test('createRun rejects with SESSION_RUN_CONFLICT when same session already has active run', async () => { + const pool = createFakePool(); + const gateway = createAgentRunGateway({ + pool, + userAuth: {}, + tkmindProxy: {}, + autoDispatch: false, + }); + + const activeRunId = crypto.randomUUID(); + const now = Date.now(); + pool.runs.set(activeRunId, { + id: activeRunId, + user_id: 'user-1', + agent_session_id: 'sess-conflict-1', + request_id: 'req-active', + status: 'running', + attempts: 1, + user_message_json: '{}', + error_message: null, + created_at: now, + updated_at: now, + started_at: now, + completed_at: null, + }); + + let conflictErr = null; + try { + await gateway.createRun('user-1', { + sessionId: 'sess-conflict-1', + requestId: 'req-conflict-2', + userMessage: { role: 'user', content: [{ type: 'text', text: 'second' }] }, + }); + } catch (err) { + conflictErr = err; + } + assert.ok(conflictErr, 'expected an error for duplicate active session run'); + assert.equal(conflictErr.code, 'SESSION_RUN_CONFLICT'); + assert.equal(conflictErr.status, 409); + + pool.runs.get(activeRunId).status = 'succeeded'; + const run3 = await gateway.createRun('user-1', { + sessionId: 'sess-conflict-1', + requestId: 'req-conflict-3', + userMessage: { role: 'user', content: [{ type: 'text', text: 'third' }] }, + }); + assert.equal(run3.requestId, 'req-conflict-3'); +}); + +test('createRun does not apply per-session conflict check for direct-chat sessions', async () => { + const pool = createFakePool(); + const gateway = createAgentRunGateway({ + pool, + userAuth: {}, + tkmindProxy: {}, + autoDispatch: false, + }); + + const activeRunId = crypto.randomUUID(); + const now = Date.now(); + pool.runs.set(activeRunId, { + id: activeRunId, + user_id: 'user-1', + agent_session_id: 'h5direct_abc', + request_id: 'req-active-dc', + status: 'running', + attempts: 1, + user_message_json: '{}', + error_message: null, + created_at: now, + updated_at: now, + started_at: now, + completed_at: null, + }); + + const run2 = await gateway.createRun('user-1', { + sessionId: 'h5direct_abc', + requestId: 'req-dc-2', + userMessage: { role: 'user', content: [{ type: 'text', text: 'hi2' }] }, + }); + + assert.equal(run2.requestId, 'req-dc-2'); +}); diff --git a/agent-run-routes.mjs b/agent-run-routes.mjs index e59f60a..fadb4a1 100644 --- a/agent-run-routes.mjs +++ b/agent-run-routes.mjs @@ -127,6 +127,13 @@ export function createPostAgentRunsHandler({ }); response.status(202).json({ run }); } catch (err) { + if (err?.code === 'SESSION_RUN_CONFLICT') { + response.status(409).json({ + message: err.message, + code: 'SESSION_RUN_CONFLICT', + }); + return; + } response.status(500).json({ message: err instanceof Error ? err.message : '创建任务失败', }); diff --git a/agent-run-routes.test.mjs b/agent-run-routes.test.mjs index 0ddefee..7368e52 100644 --- a/agent-run-routes.test.mjs +++ b/agent-run-routes.test.mjs @@ -464,6 +464,43 @@ test('POST /agent/runs rejects a session the user does not own', async () => { assert.deepEqual(res.body, { message: '无权访问该会话' }); }); +test('POST /agent/runs returns 409 when session already has active run', async () => { + const handler = createPostAgentRunsHandler({ + userAuth: { + async ownsSession() { + return true; + }, + }, + agentRunGateway: { + async createRun() { + const err = new Error('该会话有正在处理的任务，请等待完成后再发送'); + err.code = 'SESSION_RUN_CONFLICT'; + err.status = 409; + throw err; + }, + }, + }); + const res = createResponseRecorder(); + + await handler( + { + currentUser: { id: 'user-1' }, + body: { + session_id: 'session-busy', + request_id: 'req-busy', + user_message: { role: 'user', content: [{ type: 'text', text: 'hello' }] }, + }, + }, + res, + ); + + assert.equal(res.statusCode, 409); + assert.deepEqual(res.body, { + message: '该会话有正在处理的任务，请等待完成后再发送', + code: 'SESSION_RUN_CONFLICT', + }); +}); + test('POST /agent/runs validates ownership through sessionAccess when provided', async () => { const validated = []; const handler = createPostAgentRunsHandler({ diff --git a/src/hooks/useTKMindChat.ts b/src/hooks/useTKMindChat.ts index a1a0dbc..7adda6a 100644 --- a/src/hooks/useTKMindChat.ts +++ b/src/hooks/useTKMindChat.ts @@ -1280,11 +1280,13 @@ export function useTKMindChat( (value) => typeof value === 'string' && value.trim(), ); if (!text.trim() && normalizedImageUrls.length === 0) return; + // Use the ref here (not the React state) so that rapid back-to-back calls in the + // same render cycle are blocked even before the state update has been re-rendered. if ( - chatState === 'streaming' || - chatState === 'loading' || - chatState === 'connecting' || - chatState === 'waiting' + chatStateRef.current === 'streaming' || + chatStateRef.current === 'loading' || + chatStateRef.current === 'connecting' || + chatStateRef.current === 'waiting' ) return; const trimmed = text.trim(); @@ -1319,6 +1321,9 @@ export function useTKMindChat( messageHistoryLoadedCountRef.current = messagesRef.current.length; setMessages(messagesRef.current); setChatState('waiting'); + // Immediately reflect in the ref so any synchronous re-entry is blocked before + // the next React render cycle runs the useEffect that normally syncs this ref. + chatStateRef.current = 'waiting'; setError(null); setPendingTool(null); @@ -1484,7 +1489,6 @@ export function useTKMindChat( [ notifyInsufficientBalance, session, - chatState, grantedSkills, clearActiveRequestMissingTimer, subscribeToSession,