diff --git a/adapters/dingtalk/__tests__/stream-state.test.ts b/adapters/dingtalk/__tests__/stream-state.test.ts new file mode 100644 index 00000000..50a4e932 --- /dev/null +++ b/adapters/dingtalk/__tests__/stream-state.test.ts @@ -0,0 +1,53 @@ +import { describe, expect, it } from 'bun:test' +import { MessageBuffer } from '../../common/message-buffer.js' +import { finishAndResetDingTalkStreamingState, resetDingTalkStreamingState } from '../stream-state.js' + +describe('DingTalk streaming state', () => { + it('drops the active AI card stream when permission interrupts output ordering', async () => { + let resetCalled = false + const buffer = new MessageBuffer(async () => {}, 100, 1000) + const originalReset = buffer.reset.bind(buffer) + buffer.reset = () => { + resetCalled = true + originalReset() + } + + const state = { + aiCardBuffers: new Map([['chat-1', buffer]]), + streamingCards: new Map([['chat-1', Promise.resolve(null)]]), + streamingCardText: new Map([['chat-1', 'pre-permission text']]), + } + + resetDingTalkStreamingState(state, 'chat-1') + + expect(resetCalled).toBe(true) + expect(state.aiCardBuffers.has('chat-1')).toBe(false) + expect(state.streamingCards.has('chat-1')).toBe(false) + expect(state.streamingCardText.has('chat-1')).toBe(false) + }) + + it('completes the existing stream before dropping it for a permission request', async () => { + const flushed: Array<{ text: string; complete: boolean }> = [] + const buffer = new MessageBuffer( + async (text, complete) => { + flushed.push({ text, complete }) + }, + 100, + 1000, + ) + buffer.append('pre-permission text') + + const state = { + aiCardBuffers: new Map([['chat-1', buffer]]), + streamingCards: new Map([['chat-1', Promise.resolve(null)]]), + streamingCardText: new Map([['chat-1', 'already streamed']]), + } + + await finishAndResetDingTalkStreamingState(state, 'chat-1') + + expect(flushed).toEqual([{ text: 'pre-permission text', complete: true }]) + expect(state.aiCardBuffers.has('chat-1')).toBe(false) + expect(state.streamingCards.has('chat-1')).toBe(false) + expect(state.streamingCardText.has('chat-1')).toBe(false) + }) +}) diff --git a/adapters/dingtalk/index.ts b/adapters/dingtalk/index.ts index 7e73bf3b..7682da8c 100644 --- a/adapters/dingtalk/index.ts +++ b/adapters/dingtalk/index.ts @@ -44,6 +44,7 @@ import { DINGTALK_PERMISSION_CARD_CALLBACK_ROUTE, parseDingTalkPermissionCardAction, } from './permission-card.js' +import { finishAndResetDingTalkStreamingState, resetDingTalkStreamingState } from './stream-state.js' const DINGTALK_API = 'https://api.dingtalk.com' @@ -204,10 +205,7 @@ async function flushToAiCard(chatId: string, newText: string, isComplete: boolea } function clearTransientChatState(chatId: string): void { - aiCardBuffers.get(chatId)?.reset() - aiCardBuffers.delete(chatId) - streamingCards.delete(chatId) - streamingCardText.delete(chatId) + resetDingTalkStreamingState({ aiCardBuffers, streamingCards, streamingCardText }, chatId) clearPendingPermissions(chatId) const runtime = getRuntimeState(chatId) runtime.state = 'idle' @@ -388,7 +386,6 @@ async function handleServerMessage(chatId: string, msg: ServerMessage): Promise< case 'content_start': if (msg.blockType === 'text') { runtime.state = 'streaming' - void getOrCreateAiCard(chatId) } if (msg.blockType === 'tool_use') runtime.state = 'tool_executing' break @@ -428,6 +425,7 @@ async function sendPermissionRequest(chatId: string, msg: ServerMessage): Promis const runtime = getRuntimeState(chatId) runtime.pendingPermissionCount += 1 runtime.state = 'permission_pending' + await finishAndResetDingTalkStreamingState({ aiCardBuffers, streamingCards, streamingCardText }, chatId) const set = pendingPermissions.get(chatId) ?? new Set() set.add(msg.requestId) diff --git a/adapters/dingtalk/stream-state.ts b/adapters/dingtalk/stream-state.ts new file mode 100644 index 00000000..7d144c9b --- /dev/null +++ b/adapters/dingtalk/stream-state.ts @@ -0,0 +1,26 @@ +import type { MessageBuffer } from '../common/message-buffer.js' +import type { DingTalkAiCardInstance } from './ai-card.js' + +export type DingTalkStreamingState = { + aiCardBuffers: Map + streamingCards: Map> + streamingCardText: Map +} + +export function resetDingTalkStreamingState( + state: DingTalkStreamingState, + chatId: string, +): void { + state.aiCardBuffers.get(chatId)?.reset() + state.aiCardBuffers.delete(chatId) + state.streamingCards.delete(chatId) + state.streamingCardText.delete(chatId) +} + +export async function finishAndResetDingTalkStreamingState( + state: DingTalkStreamingState, + chatId: string, +): Promise { + await state.aiCardBuffers.get(chatId)?.complete() + resetDingTalkStreamingState(state, chatId) +}