fix(web): 修复运行流恢复续传序号

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
       }

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

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