From 460bd97d39762032a0c7c267386b7e423b2d3d1d Mon Sep 17 00:00:00 2001 From: Wenjie Zhang Date: Fri, 5 Jun 2026 21:02:13 +0800 Subject: [PATCH] =?UTF-8?q?fix(web):=20=E4=BF=AE=E5=A4=8D=E8=BF=90?= =?UTF-8?q?=E8=A1=8C=E6=B5=81=E6=81=A2=E5=A4=8D=E7=BB=AD=E4=BC=A0=E5=BA=8F?= =?UTF-8?q?=E5=8F=B7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- web/src/composables/useAgentRunStream.js | 49 +++++----------- .../utils/__tests__/runStreamResume.test.js | 58 +++++++++++++++++++ web/src/utils/runStreamResume.js | 54 +++++++++++++++++ 3 files changed, 126 insertions(+), 35 deletions(-) create mode 100644 web/src/utils/__tests__/runStreamResume.test.js create mode 100644 web/src/utils/runStreamResume.js diff --git a/web/src/composables/useAgentRunStream.js b/web/src/composables/useAgentRunStream.js index c0f5124d..f5b16092 100644 --- a/web/src/composables/useAgentRunStream.js +++ b/web/src/composables/useAgentRunStream.js @@ -1,47 +1,18 @@ import { unref } from 'vue' import { agentApi } from '@/apis' import { handleChatError } from '@/utils/errorHandler' +import { + compareRunSeq, + normalizeRunSeq, + resolveRunResumeAfterSeq +} from '@/utils/runStreamResume' const RUN_TERMINAL_STATUSES = new Set(['completed', 'failed', 'cancelled', 'interrupted']) const ACTIVE_RUN_STORAGE_TTL_MS = 60 * 60 * 1000 const ACTIVE_RUN_CLIENT_ID = `${Date.now()}-${Math.random().toString(36).slice(2, 10)}` -const RUN_SEQ_PATTERN = /^\d+-\d+$/ const getActiveRunStorageKey = (threadId) => `active_run:${threadId}` -const normalizeRunSeq = (value) => { - if (value === undefined || value === null) return '0-0' - const text = String(value).trim() - return RUN_SEQ_PATTERN.test(text) ? text : '0-0' -} - -const parseRunSeq = (value) => { - const text = normalizeRunSeq(value) - if (!text.includes('-')) { - return { major: 0n, minor: 0n } - } - const [majorRaw, minorRaw] = text.split('-', 2) - - try { - const major = BigInt(majorRaw || '0') - const minor = BigInt(minorRaw || '0') - return { major, minor } - } catch { - return { major: 0n, minor: 0n } - } -} - -const compareRunSeq = (incoming, current) => { - const left = parseRunSeq(incoming) - const right = parseRunSeq(current) - - if (left.major > right.major) return 1 - if (left.major < right.major) return -1 - if (left.minor > right.minor) return 1 - if (left.minor < right.minor) return -1 - return 0 -} - const getThreadIdFromObject = (value) => { if (!value || typeof value !== 'object') return '' if (typeof value.thread_id === 'string' && value.thread_id.trim()) return value.thread_id.trim() @@ -359,7 +330,14 @@ export function useAgentRunStream({ const runRes = await agentApi.getAgentRun(snapshot.run_id) const run = runRes?.run if (run && !RUN_TERMINAL_STATUSES.has(run.status)) { - await startRunStream(threadId, run.id, snapshot.last_seq || '0-0') + const afterSeq = resolveRunResumeAfterSeq({ + snapshot, + threadState: ts + }) + if (afterSeq === '0-0') { + resetOnGoingConv(threadId) + } + await startRunStream(threadId, run.id, afterSeq) return } } catch { @@ -373,6 +351,7 @@ export function useAgentRunStream({ const active = await agentApi.getThreadActiveRun(threadId) const run = active?.run if (run && !RUN_TERMINAL_STATUSES.has(run.status)) { + resetOnGoingConv(threadId) await startRunStream(threadId, run.id, '0-0') return } diff --git a/web/src/utils/__tests__/runStreamResume.test.js b/web/src/utils/__tests__/runStreamResume.test.js new file mode 100644 index 00000000..9d0f72d0 --- /dev/null +++ b/web/src/utils/__tests__/runStreamResume.test.js @@ -0,0 +1,58 @@ +import assert from 'node:assert/strict' + +import { + compareRunSeq, + hasOngoingRunChunks, + normalizeRunSeq, + resolveRunResumeAfterSeq +} from '../runStreamResume.js' + +const run = () => { + assert.equal(normalizeRunSeq(null), '0-0') + assert.equal(normalizeRunSeq('bad'), '0-0') + assert.equal(normalizeRunSeq('1700000000000-2'), '1700000000000-2') + + assert.equal(compareRunSeq('1700000000001-0', '1700000000000-9'), 1) + assert.equal(compareRunSeq('1700000000000-1', '1700000000000-9'), -1) + assert.equal(compareRunSeq('1700000000000-1', '1700000000000-1'), 0) + + const emptyThreadState = { + activeRunId: 'run-1', + runLastSeq: '1700000000005-0', + onGoingConv: { msgChunks: {} } + } + const activeThreadState = { + activeRunId: 'run-1', + runLastSeq: '1700000000005-0', + onGoingConv: { msgChunks: { 'msg-1': [{ id: 'msg-1', content: '已渲染' }] } } + } + + assert.equal(hasOngoingRunChunks(emptyThreadState), false) + assert.equal(hasOngoingRunChunks(activeThreadState), true) + + assert.equal( + resolveRunResumeAfterSeq({ + snapshot: { run_id: 'run-1', last_seq: '1700000000010-0', client_id: 'client-a' }, + threadState: emptyThreadState + }), + '0-0' + ) + assert.equal( + resolveRunResumeAfterSeq({ + snapshot: { run_id: 'run-1', last_seq: '1700000000010-0', client_id: 'client-b' }, + threadState: activeThreadState + }), + '1700000000005-0' + ) + assert.equal( + resolveRunResumeAfterSeq({ + snapshot: { run_id: 'run-2', last_seq: '1700000000010-0', client_id: 'client-a' }, + threadState: activeThreadState + }), + '0-0' + ) + + console.log('runStreamResume: all assertions passed') +} + +run() diff --git a/web/src/utils/runStreamResume.js b/web/src/utils/runStreamResume.js new file mode 100644 index 00000000..273a744c --- /dev/null +++ b/web/src/utils/runStreamResume.js @@ -0,0 +1,54 @@ +const RUN_SEQ_PATTERN = /^\d+-\d+$/ + +export const normalizeRunSeq = (value) => { + if (value === undefined || value === null) return '0-0' + const text = String(value).trim() + return RUN_SEQ_PATTERN.test(text) ? text : '0-0' +} + +const parseRunSeq = (value) => { + const text = normalizeRunSeq(value) + if (!text.includes('-')) { + return { major: 0n, minor: 0n } + } + const [majorRaw, minorRaw] = text.split('-', 2) + + try { + const major = BigInt(majorRaw || '0') + const minor = BigInt(minorRaw || '0') + return { major, minor } + } catch { + return { major: 0n, minor: 0n } + } +} + +export const compareRunSeq = (incoming, current) => { + const left = parseRunSeq(incoming) + const right = parseRunSeq(current) + + if (left.major > right.major) return 1 + if (left.major < right.major) return -1 + if (left.minor > right.minor) return 1 + if (left.minor < right.minor) return -1 + return 0 +} + +export const hasOngoingRunChunks = (threadState) => { + const msgChunks = threadState?.onGoingConv?.msgChunks + if (!msgChunks || typeof msgChunks !== 'object') return false + return Object.values(msgChunks).some((chunks) => + Array.isArray(chunks) ? chunks.length > 0 : Boolean(chunks) + ) +} + +export const resolveRunResumeAfterSeq = ({ snapshot, threadState }) => { + if (!snapshot?.run_id || !hasOngoingRunChunks(threadState)) { + return '0-0' + } + + if (threadState?.activeRunId === snapshot.run_id) { + return normalizeRunSeq(threadState.runLastSeq || snapshot.last_seq) + } + + return '0-0' +}