/** * 节流 + mutex + 冲突重刷的通用 flush 调度器 * * 这是个纯调度原语 —— 不含任何业务逻辑(发送卡片、构造 markdown 等)。 * 实际 flush 工作由构造函数注入的 doFlush 回调负责。 * * 语义: * - throttledUpdate(throttleMs) 被流式数据触发,按窗口节流 * - flush() 被 mutex 保护,相同时刻只有一个在跑 * - flush 进行中的新数据标记 needsReflush,API 结束后立即补刷 * - 长间隔(> 2000ms)后的第一次 flush 延迟 300ms 批量,避免抖动 * - complete() 后拒绝所有新 flush * * 参考实现: openclaw-lark/src/card/flush-controller.ts */ // --------------------------------------------------------------------------- // Throttle constants // --------------------------------------------------------------------------- export const THROTTLE = { /** CardKit cardElement.content() 最小间隔 —— 官方为流式设计,可高频 */ CARDKIT_MS: 100, /** im.message.patch 最小间隔 —— 严格速率限制(230020) */ PATCH_MS: 1500, /** 长间隔判定阈值。elapsed > 2000ms 触发批量模式 */ LONG_GAP_THRESHOLD_MS: 2000, /** 长间隔后第一帧的额外延迟,让文本积累更完整 */ BATCH_AFTER_GAP_MS: 300, } as const // --------------------------------------------------------------------------- // FlushController // --------------------------------------------------------------------------- export class FlushController { private flushInProgress = false private flushResolvers: Array<() => void> = [] private needsReflush = false private pendingFlushTimer: ReturnType | null = null private lastUpdateTime = 0 private isCompleted = false private _cardMessageReady = false constructor(private readonly doFlush: () => Promise) {} /** 标记完成 —— 当前 flush 跑完后不再接受新的。 */ complete(): void { this.isCompleted = true } /** 取消任何挂起的延迟 flush 计时器。 */ cancelPendingFlush(): void { if (this.pendingFlushTimer) { clearTimeout(this.pendingFlushTimer) this.pendingFlushTimer = null } } /** 等待当前正在跑的 flush 结束。没在跑则立即返回。 */ waitForFlush(): Promise { if (!this.flushInProgress) return Promise.resolve() return new Promise((resolve) => this.flushResolvers.push(resolve)) } /** * 标记卡片消息是否已发送成功,决定 flush 是否被放行。 * * 首次变 true 时同步更新 lastUpdateTime,让第一次 throttledUpdate * 看到一个小的 elapsed,匹配 openclaw 的 "card 创建完立即可刷" 行为。 */ setCardMessageReady(ready: boolean): void { this._cardMessageReady = ready if (ready) this.lastUpdateTime = Date.now() } cardMessageReady(): boolean { return this._cardMessageReady } /** * 执行一次 flush(mutex 保护 + 冲突重刷)。 * * 如果已有 flush 在跑,设置 needsReflush,当前 flush 结束后自动补一次。 */ async flush(): Promise { if (!this.cardMessageReady() || this.flushInProgress || this.isCompleted) { if (this.flushInProgress && !this.isCompleted) this.needsReflush = true return } this.flushInProgress = true this.needsReflush = false // 在 API 调用 **之前** 更新时间戳,防止并发调用者也进入 flush this.lastUpdateTime = Date.now() try { await this.doFlush() this.lastUpdateTime = Date.now() } finally { this.flushInProgress = false const resolvers = this.flushResolvers this.flushResolvers = [] for (const resolve of resolvers) resolve() // 如果 API 调用期间有新事件进来,立即补一次 flush if (this.needsReflush && !this.isCompleted && !this.pendingFlushTimer) { this.needsReflush = false this.pendingFlushTimer = setTimeout(() => { this.pendingFlushTimer = null void this.flush() }, 0) } } } /** * 节流更新入口。 * * @param throttleMs - 最小 flush 间隔。CardKit 传 THROTTLE.CARDKIT_MS, * Patch 降级路径传 THROTTLE.PATCH_MS。 */ async throttledUpdate(throttleMs: number): Promise { if (!this.cardMessageReady()) return const now = Date.now() const elapsed = now - this.lastUpdateTime if (elapsed >= throttleMs) { this.cancelPendingFlush() if (elapsed > THROTTLE.LONG_GAP_THRESHOLD_MS) { // 长间隔批量模式:工具调用 / 推理回来后的第一帧延迟 300ms // 让文本积累更完整,避免只显示一两个字 this.lastUpdateTime = now this.pendingFlushTimer = setTimeout(() => { this.pendingFlushTimer = null void this.flush() }, THROTTLE.BATCH_AFTER_GAP_MS) } else { await this.flush() } } else if (!this.pendingFlushTimer) { // 在节流窗口内 —— 延迟到窗口结束再刷 const delay = throttleMs - elapsed this.pendingFlushTimer = setTimeout(() => { this.pendingFlushTimer = null void this.flush() }, delay) } } }