Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 1e4f432754 |
@@ -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, {
|
||||
|
||||
@@ -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');
|
||||
});
|
||||
|
||||
@@ -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 : '创建任务失败',
|
||||
});
|
||||
|
||||
@@ -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({
|
||||
|
||||
@@ -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,
|
||||
|
||||
Reference in New Issue
Block a user