fix(chat): recover active requests and preserve finish order
This commit is contained in:
@@ -39,8 +39,15 @@ export function shouldPromoteSessionIdToStreaming(chatState) {
|
|||||||
* @param {number | undefined | null} status
|
* @param {number | undefined | null} status
|
||||||
* @returns {boolean}
|
* @returns {boolean}
|
||||||
*/
|
*/
|
||||||
export function shouldKeepStreamingAfterRunError(status) {
|
export function shouldKeepStreamingAfterRunError(status, message = '', code = '') {
|
||||||
return status === 0 || status === 409 || Number(status) >= 500;
|
if (status === 0 || status === 409 || Number(status) >= 500) return true;
|
||||||
|
// Goose can surface the session-level concurrency guard as a failed agent
|
||||||
|
// run (rather than an HTTP 409). The request may already be streaming and
|
||||||
|
// the session SSE is still the source of truth, so recover from the session
|
||||||
|
// instead of showing a terminal error in the chat composer.
|
||||||
|
const text = `${String(code ?? '')} ${String(message ?? '')}`.toLowerCase();
|
||||||
|
return text.includes('session already has an active request')
|
||||||
|
|| text.includes('active request. cancel it first');
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
|||||||
@@ -54,6 +54,14 @@ test('only ambiguous transport failures keep the chat streaming', () => {
|
|||||||
assert.equal(shouldKeepStreamingAfterRunError(503), true);
|
assert.equal(shouldKeepStreamingAfterRunError(503), true);
|
||||||
assert.equal(shouldKeepStreamingAfterRunError(400), false);
|
assert.equal(shouldKeepStreamingAfterRunError(400), false);
|
||||||
assert.equal(shouldKeepStreamingAfterRunError(undefined), false);
|
assert.equal(shouldKeepStreamingAfterRunError(undefined), false);
|
||||||
|
assert.equal(
|
||||||
|
shouldKeepStreamingAfterRunError(
|
||||||
|
undefined,
|
||||||
|
'Session already has an active request. Cancel it first.',
|
||||||
|
),
|
||||||
|
true,
|
||||||
|
);
|
||||||
|
assert.equal(shouldKeepStreamingAfterRunError(undefined, '后台任务失败'), false);
|
||||||
});
|
});
|
||||||
|
|
||||||
test('shouldPromoteSessionIdToStreaming skips re-streaming after Finish', () => {
|
test('shouldPromoteSessionIdToStreaming skips re-streaming after Finish', () => {
|
||||||
|
|||||||
@@ -80,3 +80,33 @@ test('mergeConversationSnapshot keeps local messages before the first and after
|
|||||||
['u0', 'u1', 'a1', 'a2'],
|
['u0', 'u1', 'a1', 'a2'],
|
||||||
);
|
);
|
||||||
});
|
});
|
||||||
|
|
||||||
|
test('mergeConversationSnapshot keeps local-only messages at their streamed position', () => {
|
||||||
|
const local = [
|
||||||
|
msg('u1', 'user', '生成页面'),
|
||||||
|
msg('a1', 'assistant', '开始分析'),
|
||||||
|
msg('a2', 'assistant', '调用工具'),
|
||||||
|
msg('a3', 'assistant', '页面完成'),
|
||||||
|
];
|
||||||
|
const server = [
|
||||||
|
msg('u1', 'user', '生成页面'),
|
||||||
|
msg('a3', 'assistant', '页面完成(服务端)'),
|
||||||
|
];
|
||||||
|
const merged = mergeConversationSnapshot(local, server);
|
||||||
|
assert.deepEqual(merged.map((message) => message.id), ['u1', 'a1', 'a2', 'a3']);
|
||||||
|
assert.equal(merged[3].content[0].text, '页面完成(服务端)');
|
||||||
|
});
|
||||||
|
|
||||||
|
test('mergeConversationSnapshot preserves multiple local messages before the next server anchor', () => {
|
||||||
|
const local = [
|
||||||
|
msg('u1', 'user', '任务'),
|
||||||
|
msg('a1', 'assistant', '步骤一'),
|
||||||
|
msg('a2', 'assistant', '步骤二'),
|
||||||
|
msg('a3', 'assistant', '完成'),
|
||||||
|
];
|
||||||
|
const server = [msg('u1', 'user', '任务'), msg('a3', 'assistant', '完成')];
|
||||||
|
assert.deepEqual(
|
||||||
|
mergeConversationSnapshot(local, server).map((message) => message.id),
|
||||||
|
['u1', 'a1', 'a2', 'a3'],
|
||||||
|
);
|
||||||
|
});
|
||||||
|
|||||||
@@ -1618,7 +1618,7 @@ export function useTKMindChat(
|
|||||||
if (
|
if (
|
||||||
activeSessionId &&
|
activeSessionId &&
|
||||||
err instanceof ApiError &&
|
err instanceof ApiError &&
|
||||||
shouldKeepStreamingAfterRunError(err.status)
|
shouldKeepStreamingAfterRunError(err.status, err.message, err.code)
|
||||||
) {
|
) {
|
||||||
subscribeToSession(activeSessionId);
|
subscribeToSession(activeSessionId);
|
||||||
setChatState('streaming');
|
setChatState('streaming');
|
||||||
@@ -1626,6 +1626,23 @@ export function useTKMindChat(
|
|||||||
scheduleReplyRecoverySync(activeSessionId, submitToken);
|
scheduleReplyRecoverySync(activeSessionId, submitToken);
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
if (
|
||||||
|
activeSessionId &&
|
||||||
|
shouldKeepStreamingAfterRunError(
|
||||||
|
undefined,
|
||||||
|
err instanceof Error ? err.message : String(err),
|
||||||
|
err instanceof ApiError ? err.code : '',
|
||||||
|
)
|
||||||
|
) {
|
||||||
|
// Goose may report its session concurrency guard as a failed run
|
||||||
|
// message instead of an HTTP 409. Reattach to the session stream
|
||||||
|
// and reconcile the snapshot; do not strand the composer in error.
|
||||||
|
subscribeToSession(activeSessionId);
|
||||||
|
setChatState('streaming');
|
||||||
|
setError(null);
|
||||||
|
scheduleReplyRecoverySync(activeSessionId, submitToken);
|
||||||
|
return;
|
||||||
|
}
|
||||||
if (session && activeSessionId) setSessions((prev) => touchSession(prev, activeSessionId!, -1));
|
if (session && activeSessionId) setSessions((prev) => touchSession(prev, activeSessionId!, -1));
|
||||||
if (err instanceof ApiError && err.status === 402) {
|
if (err instanceof ApiError && err.status === 402) {
|
||||||
notifyInsufficientBalance();
|
notifyInsufficientBalance();
|
||||||
|
|||||||
Reference in New Issue
Block a user