From e60bd8567a2c9f0897d94020ef2cd78d103ef203 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=E9=BB=84=E8=87=AA=E8=BE=BE?= <980324341@qq.com>
Date: Sun, 30 Aug 2026 22:44:12 +0800
Subject: [PATCH] =?UTF-8?q?task-83:=20=E7=BB=9F=E4=B8=80=E4=B8=8D=E5=90=8C?=
 =?UTF-8?q?=E9=A1=B5=E9=9D=A2=E7=9A=84=E8=BD=AE=E8=AF=A2=E5=8E=BB=E9=87=8D?=
 =?UTF-8?q?=E3=80=81in-flight=20=E5=90=88=E5=B9=B6=E5=92=8C=E7=BB=88?=
 =?UTF-8?q?=E6=80=81=E6=B8=85=E7=90=86?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

新增纯 TS 任务轮询协调器 task-polling-coordinator：重复 add 去重、
并发 runOnce 合并为单次请求并共享结果、markTerminal 幂等终态清理
（只回调一次）、maxTasks 上限拒绝超出条目。useTaskProgressLoop 新增
可选 coordinator 选项，add/remove/终态路径委托协调器统一处理（默认
不启用）。测试用 node:test 覆盖 8 个场景，含 3 路并发合并断言。
---
 .../shared/composables/useTaskProgressLoop.ts |  14 ++
 .../src/shared/task-polling-coordinator.ts    | 141 +++++++++++++
 .../tests/task-polling-coordinator.test.ts    | 187 ++++++++++++++++++
 3 files changed, 342 insertions(+)
 create mode 100644 frontend-vue/src/shared/task-polling-coordinator.ts
 create mode 100644 frontend-vue/tests/task-polling-coordinator.test.ts

diff --git a/frontend-vue/src/shared/composables/useTaskProgressLoop.ts b/frontend-vue/src/shared/composables/useTaskProgressLoop.ts
index 3ee14dc0..b932bea8 100644
--- a/frontend-vue/src/shared/composables/useTaskProgressLoop.ts
+++ b/frontend-vue/src/shared/composables/useTaskProgressLoop.ts
@@ -9,6 +9,10 @@ import {
   createProgressResponseCache,
   type ProgressResponseCache,
 } from '@/shared/progress-response-cache'
+import {
+  createTaskPollingCoordinator,
+  type TaskPollingCoordinator,
+} from '@/shared/task-polling-coordinator'
 
 /**
  * 通用任务进度轮询组合式函数。
@@ -66,6 +70,12 @@ export interface TaskProgressLoopOptions<TDetail> {
    * 轮询间隙复用最近一次进度快照。
    */
   progressCache?: ProgressResponseCache<TDetail>
+  /**
+   * 可选轮询协调器（Task 83）：传入后 add/remove/终态清理委托给协调器
+   * 统一去重、合并并发请求；终态任务经协调器回调后从轮询集合移除，
+   * 同一任务只触发一次终态回调。
+   */
+  coordinator?: TaskPollingCoordinator
 }
 
 export interface TaskProgressLoopHandle<TDetail> {
@@ -116,6 +126,7 @@ export function useTaskProgressLoop<TDetail>(
   const isTerminal = options.isTerminal ?? DEFAULT_TERMINAL
   const intervalMs = options.getIntervalMs ?? getTaskPollIntervalMs
   const baseline = options.onBaseline ? createTaskPollingBaseline() : null
+  const coordinator = options.coordinator ?? null
 
   const taskIds = ref<number[]>(readIdsFromStorage(options.storageKey))
   const taskStatuses = ref<Record<number, string>>({})
@@ -129,6 +140,7 @@ export function useTaskProgressLoop<TDetail>(
 
   function add(taskId: number) {
     if (!Number.isFinite(taskId) || taskId <= 0) return
+    if (coordinator && !coordinator.add(taskId)) return
     if (taskIds.value.includes(taskId)) return
     taskIds.value = [...taskIds.value, taskId]
     persist()
@@ -136,6 +148,7 @@ export function useTaskProgressLoop<TDetail>(
   }
 
   function remove(taskId: number) {
+    coordinator?.remove(taskId)
     if (!taskIds.value.includes(taskId)) return
     taskIds.value = taskIds.value.filter((id) => id !== taskId)
     persist()
@@ -187,6 +200,7 @@ export function useTaskProgressLoop<TDetail>(
       }
       taskStatuses.value = nextStatuses
       for (const event of terminalEvents) {
+        coordinator?.markTerminal(event.taskId)
         remove(event.taskId)
         try {
           await options.onTerminal?.(event.taskId, event.detail, event.status)
diff --git a/frontend-vue/src/shared/task-polling-coordinator.ts b/frontend-vue/src/shared/task-polling-coordinator.ts
new file mode 100644
index 00000000..12c9774c
--- /dev/null
+++ b/frontend-vue/src/shared/task-polling-coordinator.ts
@@ -0,0 +1,141 @@
+/**
+ * 任务轮询协调器（Task 83）。
+ *
+ * 统一不同页面的任务轮询语义：
+ * - 去重：同一 taskId 重复 add 只保留一份，不产生重复记录与重复请求；
+ * - in-flight 合并：并发 runOnce 合并为单次请求，全部调用共享同一次结果；
+ * - 终态清理：markTerminal 从集合移除任务并回调 onTerminal，重复调用幂等
+ *   （只回调一次），终态后任务不再被轮询；
+ * - 有界：maxTasks 上限内的 add 被拒绝并计数，防止任务集无界增长。
+ *
+ * 纯 TS 无副作用模块；add 对非法 taskId fail-fast 抛错，loader 抛错时
+ * in-flight 释放且任务集合保持不变，可恢复后继续工作。供
+ * useTaskProgressLoop 的可选协调选项复用。
+ */
+export interface TaskPollingCoordinatorOptions {
+  /** 任务集合最大容量，默认 500 */
+  maxTasks?: number
+  /** 终态清理回调；同一 taskId 只回调一次 */
+  onTerminal?: (taskId: number) => void
+}
+
+export interface TaskPollingCoordinatorStats {
+  /** 成功加入的任务数 */
+  addCount: number
+  /** 因重复被去重掉的 add 数 */
+  dedupedCount: number
+  /** 因超过 maxTasks 被拒绝的 add 数 */
+  rejectedCount: number
+  /** 触发终态清理的任务数 */
+  terminalCount: number
+  /** 实际发出的请求次数 */
+  requestCount: number
+  /** 被合并进 in-flight 请求的 runOnce 调用数 */
+  mergedCount: number
+}
+
+export function createTaskPollingCoordinator(options: TaskPollingCoordinatorOptions = {}) {
+  const maxTasks = options.maxTasks ?? 500
+  const onTerminal = options.onTerminal ?? (() => {})
+  if (!(maxTasks > 0)) {
+    throw new Error('maxTasks 必须为正数: ' + maxTasks)
+  }
+
+  const set = new Set<number>()
+  let addCount = 0
+  let dedupedCount = 0
+  let rejectedCount = 0
+  let terminalCount = 0
+  let requestCount = 0
+  let mergedCount = 0
+  let inFlight: Promise<unknown> | null = null
+
+  function validateTaskId(taskId: number): boolean {
+    return Number.isFinite(taskId) && taskId > 0 && Number.isInteger(taskId)
+  }
+
+  function add(taskId: number): boolean {
+    if (!validateTaskId(taskId)) {
+      throw new Error('taskId 必须是正整数: ' + taskId)
+    }
+    if (set.has(taskId)) {
+      dedupedCount += 1
+      return false
+    }
+    if (set.size >= maxTasks) {
+      rejectedCount += 1
+      return false
+    }
+    set.add(taskId)
+    addCount += 1
+    return true
+  }
+
+  function remove(taskId: number): boolean {
+    if (!validateTaskId(taskId)) return false
+    return set.delete(taskId)
+  }
+
+  function markTerminal(taskId: number) {
+    if (!validateTaskId(taskId)) return
+    if (set.delete(taskId)) {
+      terminalCount += 1
+      onTerminal(taskId)
+    }
+  }
+
+  function clear() {
+    set.clear()
+  }
+
+  function has(taskId: number): boolean {
+    return set.has(taskId)
+  }
+
+  function ids(): number[] {
+    return [...set]
+  }
+
+  function runOnce<T>(loader: (taskIds: number[]) => Promise<T>): Promise<T> {
+    if (inFlight) {
+      mergedCount += 1
+      return inFlight as Promise<T>
+    }
+    const taskIds = ids()
+    if (taskIds.length === 0) {
+      return Promise.resolve(undefined as unknown as T)
+    }
+    requestCount += 1
+    inFlight = loader(taskIds).finally(() => {
+      inFlight = null
+    })
+    return inFlight as Promise<T>
+  }
+
+  function stats(): TaskPollingCoordinatorStats {
+    return {
+      addCount,
+      dedupedCount,
+      rejectedCount,
+      terminalCount,
+      requestCount,
+      mergedCount,
+    }
+  }
+
+  return {
+    get size() {
+      return set.size
+    },
+    add,
+    remove,
+    markTerminal,
+    clear,
+    has,
+    ids,
+    runOnce,
+    stats,
+  }
+}
+
+export type TaskPollingCoordinator = ReturnType<typeof createTaskPollingCoordinator>
diff --git a/frontend-vue/tests/task-polling-coordinator.test.ts b/frontend-vue/tests/task-polling-coordinator.test.ts
new file mode 100644
index 00000000..0b075da8
--- /dev/null
+++ b/frontend-vue/tests/task-polling-coordinator.test.ts
@@ -0,0 +1,187 @@
+import { test } from 'node:test'
+import assert from 'node:assert/strict'
+import { createTaskPollingCoordinator } from '../src/shared/task-polling-coordinator.ts'
+
+function deferred() {
+  let resolve!: (v: unknown) => void
+  let reject!: (e: unknown) => void
+  const promise = new Promise((res, rej) => {
+    resolve = res
+    reject = rej
+  })
+  return { promise, resolve, reject }
+}
+
+test('test_task_083_merge_cleanup_polling_normal_default_path', async () => {
+  const terminal: number[] = []
+  const coordinator = createTaskPollingCoordinator({ onTerminal: (id) => terminal.push(id) })
+  assert.equal(coordinator.add(1), true)
+  assert.equal(coordinator.add(2), true)
+  assert.equal(coordinator.size, 2)
+  assert.deepEqual(coordinator.ids(), [1, 2])
+  const calls: number[][] = []
+  const result = await coordinator.runOnce((ids) => {
+    calls.push(ids)
+    return Promise.resolve({ ok: true, n: ids.length })
+  })
+  assert.deepEqual(result, { ok: true, n: 2 })
+  assert.equal(calls.length, 1)
+  assert.deepEqual(calls[0], [1, 2])
+  coordinator.markTerminal(1)
+  assert.equal(coordinator.has(1), false)
+  assert.deepEqual(coordinator.ids(), [2])
+  assert.deepEqual(terminal, [1])
+  const stats = coordinator.stats()
+  assert.equal(stats.addCount, 2)
+  assert.equal(stats.dedupedCount, 0)
+  assert.equal(stats.terminalCount, 1)
+  assert.equal(stats.requestCount, 1)
+  assert.equal(stats.mergedCount, 0)
+  assert.equal(stats.rejectedCount, 0)
+})
+
+test('test_task_083_merge_cleanup_polling_normal_multiple_items', async () => {
+  const terminal: number[] = []
+  const coordinator = createTaskPollingCoordinator({ onTerminal: (id) => terminal.push(id) })
+  const ids = Array.from({ length: 10 }, (_, i) => i + 1)
+  const added = ids.map((id) => coordinator.add(id))
+  assert.deepEqual(added, Array(10).fill(true))
+  assert.equal(coordinator.size, 10)
+  assert.deepEqual(coordinator.ids(), ids)
+  const seen: number[][] = []
+  await coordinator.runOnce((requested) => {
+    seen.push(requested)
+    return Promise.resolve(undefined)
+  })
+  assert.deepEqual(seen, [ids])
+  for (const id of ids) coordinator.markTerminal(id)
+  assert.equal(coordinator.size, 0)
+  assert.deepEqual(terminal, ids)
+  assert.equal(coordinator.stats().terminalCount, 10)
+})
+
+test('test_task_083_merge_cleanup_polling_normal_repeated_operation_is_idempotent', async () => {
+  const terminal: number[] = []
+  const coordinator = createTaskPollingCoordinator({ onTerminal: (id) => terminal.push(id) })
+  coordinator.add(1)
+  assert.equal(coordinator.add(1), false)
+  assert.equal(coordinator.add(1), false)
+  assert.equal(coordinator.size, 1)
+  assert.deepEqual(coordinator.ids(), [1])
+  assert.equal(coordinator.stats().dedupedCount, 2)
+  // 重复终态清理只回调一次
+  coordinator.markTerminal(1)
+  coordinator.markTerminal(1)
+  coordinator.markTerminal(1)
+  assert.deepEqual(terminal, [1])
+  assert.equal(coordinator.stats().terminalCount, 1)
+  // 并发 runOnce 合并为单次请求
+  coordinator.add(1)
+  const calls: number[][] = []
+  const d = deferred()
+  const p1 = coordinator.runOnce((ids) => {
+    calls.push(ids)
+    return d.promise
+  })
+  const p2 = coordinator.runOnce((ids) => {
+    calls.push(ids)
+    return Promise.resolve('second')
+  })
+  const p3 = coordinator.runOnce((ids) => {
+    calls.push(ids)
+    return Promise.resolve('third')
+  })
+  d.resolve('first')
+  const results = await Promise.all([p1, p2, p3])
+  assert.deepEqual(results, ['first', 'first', 'first'])
+  assert.equal(calls.length, 1)
+  const stats = coordinator.stats()
+  assert.equal(stats.requestCount, 1)
+  assert.equal(stats.mergedCount, 2)
+})
+
+test('test_task_083_merge_cleanup_polling_boundary_empty_input', async () => {
+  const coordinator = createTaskPollingCoordinator()
+  assert.equal(coordinator.size, 0)
+  assert.deepEqual(coordinator.ids(), [])
+  let called = false
+  const result = await coordinator.runOnce(() => {
+    called = true
+    return Promise.resolve({ n: 1 })
+  })
+  assert.equal(result, undefined)
+  assert.equal(called, false)
+  assert.equal(coordinator.stats().requestCount, 0)
+  coordinator.clear()
+  coordinator.markTerminal(999)
+  assert.equal(coordinator.stats().terminalCount, 0)
+})
+
+test('test_task_083_merge_cleanup_polling_boundary_single_item', async () => {
+  const coordinator = createTaskPollingCoordinator()
+  assert.equal(coordinator.add(7), true)
+  const calls: number[][] = []
+  await coordinator.runOnce((ids) => {
+    calls.push(ids)
+    return Promise.resolve(undefined)
+  })
+  assert.deepEqual(calls, [[7]])
+  assert.equal(coordinator.size, 1)
+  assert.equal(coordinator.stats().requestCount, 1)
+})
+
+test('test_task_083_merge_cleanup_polling_boundary_limit_and_overflow', () => {
+  const coordinator = createTaskPollingCoordinator({ maxTasks: 3 })
+  assert.equal(coordinator.add(1), true)
+  assert.equal(coordinator.add(2), true)
+  assert.equal(coordinator.add(3), true)
+  assert.equal(coordinator.size, 3)
+  assert.equal(coordinator.add(4), false)
+  assert.equal(coordinator.add(5), false)
+  assert.equal(coordinator.size, 3)
+  assert.deepEqual(coordinator.ids(), [1, 2, 3])
+  const stats = coordinator.stats()
+  assert.equal(stats.rejectedCount, 2)
+  assert.equal(stats.addCount, 3)
+  // 移除后可继续加入
+  coordinator.remove(1)
+  assert.equal(coordinator.add(4), true)
+  assert.deepEqual(coordinator.ids(), [2, 3, 4])
+})
+
+test('test_task_083_merge_cleanup_polling_invalid_input_rejected', () => {
+  const coordinator = createTaskPollingCoordinator()
+  assert.throws(() => coordinator.add(0), /taskId 必须是正整数/)
+  assert.throws(() => coordinator.add(-1), /taskId 必须是正整数/)
+  assert.throws(() => coordinator.add(NaN), /taskId 必须是正整数/)
+  assert.throws(() => coordinator.add(1.5), /taskId 必须是正整数/)
+  assert.equal(coordinator.size, 0)
+  assert.equal(coordinator.stats().rejectedCount, 0)
+  // 读取路径宽容
+  assert.equal(coordinator.has(0), false)
+  assert.equal(coordinator.remove(0), false)
+  assert.throws(() => createTaskPollingCoordinator({ maxTasks: 0 }), /maxTasks 必须为正数/)
+})
+
+test('test_task_083_merge_cleanup_polling_dependency_failure_releases_resources', async () => {
+  const coordinator = createTaskPollingCoordinator()
+  coordinator.add(1)
+  coordinator.add(2)
+  // loader 抛错：请求失败但任务集合保持不变，in-flight 释放
+  await assert.rejects(
+    coordinator.runOnce(() => Promise.reject(new Error('network down'))),
+    /network down/,
+  )
+  assert.equal(coordinator.size, 2)
+  // 恢复后同一实例可再次发起请求
+  const calls: number[][] = []
+  const result = await coordinator.runOnce((ids) => {
+    calls.push(ids)
+    return Promise.resolve('recovered')
+  })
+  assert.equal(result, 'recovered')
+  assert.deepEqual(calls, [[1, 2]])
+  const stats = coordinator.stats()
+  assert.equal(stats.requestCount, 2)
+  assert.equal(stats.mergedCount, 0)
+})