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

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

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
john
2026-07-07 11:48:59 +08:00
parent ac520d8ae5
commit 1e4f432754
5 changed files with 162 additions and 5 deletions
+92
View File
@@ -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');
});