fix(chat): 阻止同会话并发 Agent Run 导致消息错乱

前端 submit 改用 chatStateRef 同步拦截连发；后端 createRun 检测同 session 活跃任务并返回 409，避免多条回复交错落库。

Co-authored-by: Cursor <cursoragent@cursor.com>

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, {

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');
+});

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 : '创建任务失败',
       });

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({

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' ||
+        chatStateRef.current === 'streaming' ||
-        chatState === 'loading' ||
+        chatStateRef.current === 'loading' ||
-        chatState === 'connecting' ||
+        chatStateRef.current === 'connecting' ||
-        chatState === 'waiting'
+        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,