From ef650accf8cf3ddb3796aa6946d105307a44d76f Mon Sep 17 00:00:00 2001 From: qqqys <266654365+qqqys@users.noreply.github.com> Date: Tue, 1 Sep 2026 17:00:59 +0800 Subject: [PATCH 1/5] fix(dws): isolate direct message ingestion --- packages/channels/dws/src/dws-channel.test.ts | 215 ++++++++++++++++- packages/channels/dws/src/dws-channel.ts | 223 ++++++++++++++---- packages/channels/dws/src/dws-client.test.ts | 104 ++++++++ packages/channels/dws/src/dws-client.ts | 30 ++- packages/channels/dws/src/index.ts | 2 + 5 files changed, 520 insertions(+), 54 deletions(-) diff --git a/packages/channels/dws/src/dws-channel.test.ts b/packages/channels/dws/src/dws-channel.test.ts index b0c1f28e907..dac7ebcb77b 100644 --- a/packages/channels/dws/src/dws-channel.test.ts +++ b/packages/channels/dws/src/dws-channel.test.ts @@ -22,6 +22,7 @@ import { type DwsClientLike, type DwsCommandRunner, type DwsIdentity, + type DwsImMessageResult, type DwsImMessage, type DwsImSource, type DwsImTarget, @@ -152,7 +153,7 @@ class FakeSubscription implements DwsEventSubscription { interface FakeStream { source: DwsImSource; - onMessage: (message: DwsImMessage) => void | Promise; + onMessage: (message: DwsImMessage) => DwsImMessageResult; onError: (error: Error) => void; subscription: FakeSubscription; } @@ -227,7 +228,7 @@ class FakeDwsClient implements DwsClientLike { async subscribeToIm( source: DwsImSource, - onMessage: (message: DwsImMessage) => void | Promise, + onMessage: (message: DwsImMessage) => DwsImMessageResult, onError: (error: Error) => void, ): Promise { const subscription = new FakeSubscription(); @@ -238,7 +239,9 @@ class FakeDwsClient implements DwsClientLike { async emit(sourceIndex: number, event: DwsImMessage): Promise { const stream = this.streams[sourceIndex]; if (!stream) throw new Error(`Missing fake stream ${sourceIndex}.`); - await stream.onMessage(event); + const result = stream.onMessage(event); + if (result && 'completed' in result) await result.completed; + else await result; } } @@ -321,6 +324,36 @@ class TestableDwsChannel extends DwsChannel { return this.cursor.mentionWatermark; } + pendingMessageIds(): string[] { + return (this.cursor.pendingMessages ?? []).map( + ({ message }) => message.messageId, + ); + } + + seedPendingMessages(count: number): void { + this.cursor.pendingMessages = Array.from( + { length: count }, + (_unused, index) => ({ + source: { kind: 'direct' } as const, + message: message( + 'user_im_message_receive_o2o_all', + `parked-${index}`, + `request ${index}`, + { conversationId: 'conversation-capacity' }, + ), + }), + ); + this.saveCursor(); + } + + releasePendingMessage(conversationId: string, messageId: string): void { + const removePendingMessage = ( + this as unknown as { removePendingMessage(key: string): void } + ).removePendingMessage.bind(this); + removePendingMessage(`${conversationId}\0${messageId}`); + this.saveCursor(); + } + resolveSession(): Promise { return this.router.resolve(this.name, 'alice', 'doc-1', 'comment-1'); } @@ -1318,6 +1351,131 @@ describe('DwsChannel', () => { ); }); + it('does not let one direct conversation block another', async () => { + const client = new FakeDwsClient(); + const channel = await readyChannel(client); + let releaseFirst!: () => void; + const firstBlocked = new Promise((resolve) => { + releaseFirst = resolve; + }); + let releaseSecond!: () => void; + const secondBlocked = new Promise((resolve) => { + releaseSecond = resolve; + }); + const started: string[] = []; + channel.inboundHandler = async (envelope) => { + started.push(envelope.messageId); + if (envelope.messageId === 'conversation-a') await firstBlocked; + if (envelope.messageId === 'conversation-b') await secondBlocked; + channel.inbound.push(envelope); + }; + + const firstDelivery = client.emit( + 1, + message( + 'user_im_message_receive_o2o_all', + 'conversation-a', + 'first request', + { conversationId: 'conversation-a' }, + ), + ); + await vi.waitFor(() => expect(started).toEqual(['conversation-a'])); + expect(channel.pendingMessageIds()).toEqual(['conversation-a']); + + const secondDelivery = client.emit( + 1, + message( + 'user_im_message_receive_o2o_all', + 'conversation-b', + 'second request', + { conversationId: 'conversation-b' }, + ), + ); + + await vi.waitFor(() => + expect(started).toEqual(['conversation-a', 'conversation-b']), + ); + expect(channel.pendingMessageIds()).toEqual([ + 'conversation-a', + 'conversation-b', + ]); + releaseFirst(); + releaseSecond(); + await Promise.all([firstDelivery, secondDelivery]); + await vi.waitFor(() => expect(channel.inbound).toHaveLength(2)); + expect(channel.pendingMessageIds()).toEqual([]); + }); + + it('preserves direct-message order within one conversation', async () => { + const client = new FakeDwsClient(); + const { bridge } = await readyPolicyChannel( + client, + makeConfig({ dispatchMode: 'followup' }), + ); + let releaseFirst!: () => void; + const firstBlocked = new Promise((resolve) => { + releaseFirst = resolve; + }); + let promptCount = 0; + (bridge.prompt as ReturnType).mockImplementation(async () => { + promptCount += 1; + if (promptCount === 1) await firstBlocked; + return 'response'; + }); + + const firstDelivery = client.emit( + 1, + message( + 'user_im_message_receive_o2o_all', + 'conversation-a-1', + 'first request', + { conversationId: 'conversation-a' }, + ), + ); + await vi.waitFor(() => expect(bridge.prompt).toHaveBeenCalledOnce()); + const secondDelivery = client.emit( + 1, + message( + 'user_im_message_receive_o2o_all', + 'conversation-a-2', + 'second request', + { conversationId: 'conversation-a' }, + ), + ); + + await new Promise((resolve) => setTimeout(resolve, 0)); + expect(bridge.prompt).toHaveBeenCalledOnce(); + releaseFirst(); + await Promise.all([firstDelivery, secondDelivery]); + expect(bridge.prompt).toHaveBeenCalledTimes(2); + }); + + it('applies backpressure instead of evicting an admitted message', async () => { + const client = new FakeDwsClient(); + const channel = await readyChannel(client); + channel.seedPendingMessages(5_000); + + const delivery = client.emit( + 1, + message('user_im_message_receive_o2o_all', 'new-request', 'request', { + conversationId: 'conversation-new', + }), + ); + await new Promise((resolve) => setTimeout(resolve, 0)); + + expect(channel.inbound).toEqual([]); + expect(channel.pendingMessageIds()).toHaveLength(5_000); + expect(channel.pendingMessageIds()[0]).toBe('parked-0'); + expect(channel.pendingMessageIds()).not.toContain('new-request'); + + channel.releasePendingMessage('conversation-capacity', 'parked-0'); + await delivery; + expect(channel.inbound.map(({ messageId }) => messageId)).toEqual([ + 'new-request', + ]); + expect(channel.pendingMessageIds()).toHaveLength(4_999); + }); + it('turns a document mention notification into a document task and replies to its comment', async () => { const client = new FakeDwsClient(); const channel = await readyChannel(client); @@ -1828,6 +1986,42 @@ describe('DwsChannel', () => { ]); }); + it('admits a direct-message history page without waiting for a turn', async () => { + const client = new FakeDwsClient(); + const channel = await readyChannel(client); + let releaseFirst!: () => void; + const firstBlocked = new Promise((resolve) => { + releaseFirst = resolve; + }); + const started: string[] = []; + channel.inboundHandler = async (envelope) => { + started.push(envelope.messageId); + if (envelope.messageId === 'history-a') await firstBlocked; + channel.inbound.push(envelope); + }; + const before = channel.notificationWatermark(); + const eventTime = Date.now(); + client.directMessages = [ + message('user_im_message_receive_o2o_all', 'history-a', 'first request', { + conversationId: 'conversation-a', + eventTime, + }), + message( + 'user_im_message_receive_o2o_all', + 'history-b', + 'second request', + { conversationId: 'conversation-b', eventTime }, + ), + ]; + + await channel.poll(); + + await vi.waitFor(() => expect(started).toEqual(['history-a', 'history-b'])); + expect(channel.notificationWatermark()).toBeGreaterThanOrEqual(before!); + releaseFirst(); + await vi.waitFor(() => expect(channel.inbound).toHaveLength(2)); + }); + // R1-7: every history window re-opens at `watermark - 5s`, so every // live-dispatched direct message is re-fetched by a later poll. The // processed-key guard is the only thing standing between that refetch and @@ -1916,20 +2110,21 @@ describe('DwsChannel', () => { await channel.poll(); - expect(channel.inboundAttempts).toBe(2); + await vi.waitFor(() => expect(channel.inboundAttempts).toBe(2)); const logged = stderr.mock.calls.map((call) => String(call[0])).join(''); expect(logged).toContain('DWS message turn failed (attempt 1/5)'); - expect(logged).toContain('parked for retry'); expect(logged).not.toContain('failed to poll DWS direct-message history'); channel.inboundError = undefined; await channel.poll(); - expect(channel.inboundAttempts).toBe(4); - expect(channel.inbound.map((envelope) => envelope.messageId)).toEqual([ - 'failing-first', - 'waiting-second', - ]); + await vi.waitFor(() => expect(channel.inboundAttempts).toBe(4)); + await vi.waitFor(() => + expect(channel.inbound.map((envelope) => envelope.messageId)).toEqual([ + 'failing-first', + 'waiting-second', + ]), + ); } finally { stderr.mockRestore(); } diff --git a/packages/channels/dws/src/dws-channel.ts b/packages/channels/dws/src/dws-channel.ts index 6da936ea2db..ca282d01017 100644 --- a/packages/channels/dws/src/dws-channel.ts +++ b/packages/channels/dws/src/dws-channel.ts @@ -23,6 +23,7 @@ import { DwsClient, DwsCommandError, type DwsClientLike, + type DwsImDispatch, type DwsImMessage, type DwsImSource, type DwsImTarget, @@ -518,6 +519,8 @@ export class DwsChannel extends PollingChannelBase { private readonly endReactionKeys = new Set(); private readonly notifiedSenderPairingNotifications = new Set(); private readonly processingMessages = new Map>(); + private readonly queuedDirectMessages = new Map>(); + private readonly pendingMessageCapacityWaiters = new Set<() => void>(); private pollAbortController = new AbortController(); private lifecycleGeneration = 0; private connectionStartedAt = 0; @@ -825,6 +828,9 @@ export class DwsChannel extends PollingChannelBase { this.cleanupReaction(key, 'disconnect reaction removal'); } this.sessionReactionKeys.clear(); + this.queuedDirectMessages.clear(); + for (const resolve of this.pendingMessageCapacityWaiters) resolve(); + this.pendingMessageCapacityWaiters.clear(); this.stopPollLoop(); for (const state of this.imStates) { if (state.retryTimer) clearTimeout(state.retryTimer); @@ -1124,7 +1130,7 @@ export class DwsChannel extends PollingChannelBase { ); for (const message of mentions.messages) { if (signal.aborted || !this.connected) return; - await this.handleImMessage({ kind: 'at' }, message, true); + await this.receiveImMessage({ kind: 'at' }, message, true); } if (mentions.nextCursor) { this.cursor.mentionCheckpoint = { @@ -1172,7 +1178,10 @@ export class DwsChannel extends PollingChannelBase { // shared retry budget twice per poll. if (this.hasPendingMessage(key)) continue; try { - await this.handleImMessage({ kind: 'direct' }, message, true); + const dispatch = this.receiveDirectMessage(message, true); + await (parseDocumentMentionNotification(message.content.trim()) + ? dispatch.completed + : dispatch.admitted); } catch (error) { // A failed plain direct message was just parked for replay, so the // page can keep moving. An unparked failure — a document @@ -1370,7 +1379,9 @@ export class DwsChannel extends PollingChannelBase { (message) => { state.lastError = undefined; state.restartAttempts = 0; - return this.handleImMessage(state.source, message); + return state.source.kind === 'direct' + ? this.receiveDirectMessage(message) + : this.receiveImMessage(state.source, message); }, (error) => { if (error instanceof DwsEventProcessError) state.lastError = error; @@ -1493,11 +1504,15 @@ export class DwsChannel extends PollingChannelBase { ); } - private async handleImMessage( + private async receiveImMessage( source: DwsImSource, message: DwsImMessage, fromHistory = false, ): Promise { + if (source.kind === 'direct') { + await this.receiveDirectMessage(message, fromHistory).completed; + return; + } if (!this.connected) return; if (this.isSelfMessage(message)) { this.markProcessedMessage(messageKey(message)); @@ -1509,11 +1524,57 @@ export class DwsChannel extends PollingChannelBase { message.eventTime !== undefined && message.eventTime < this.connectionStartedAt - 5_000 ) { - if (source.kind !== 'direct') { - this.markProcessedMessage(messageKey(message)); - this.saveCursor(); - return; - } + this.markProcessedMessage(messageKey(message)); + this.saveCursor(); + return; + } + if ( + source.kind === 'group-all' && + (this.config.groups[message.conversationId]?.requireMention ?? + this.config.groups['*']?.requireMention ?? + true) + ) { + return; + } + if ( + (source.kind === 'group' || source.kind === 'group-all') && + this.config.groupPolicy === 'pairing' && + !this.groupGate.isGroupApproved(message.conversationId) + ) { + return; + } + await this.dispatchImMessage(source, message, messageKey(message)); + } + + private receiveDirectMessage( + message: DwsImMessage, + fromHistory = false, + ): DwsImDispatch { + const admission = this.admitReceivedDirectMessage(message, fromHistory); + const completed = admission.then(({ completion }) => completion); + void completed.catch(() => undefined); + return { + admitted: admission.then(() => undefined), + completed, + }; + } + + private async admitReceivedDirectMessage( + message: DwsImMessage, + fromHistory: boolean, + ): Promise<{ completion: Promise }> { + if (!this.connected) return { completion: Promise.resolve() }; + const key = messageKey(message); + if (this.isSelfMessage(message)) { + this.markProcessedMessage(key); + this.saveCursor(); + return { completion: Promise.resolve() }; + } + if ( + !fromHistory && + message.eventTime !== undefined && + message.eventTime < this.connectionStartedAt - 5_000 + ) { // A replayed direct message is left UNMARKED on purpose, for history // polling to pick up. That only works if polling will ever look // that far back: on a fresh cursor `notificationWatermark` starts at @@ -1540,24 +1601,63 @@ export class DwsChannel extends PollingChannelBase { `[Channel:${this.name}] parked a stale direct message for history polling and pulled the watermark back to ${message.eventTime}: ${sanitizeLogText(message.messageId, 120)}\n`, ); this.saveCursor(); - return; - } - if ( - source.kind === 'group-all' && - (this.config.groups[message.conversationId]?.requireMention ?? - this.config.groups['*']?.requireMention ?? - true) - ) { - return; + return { completion: Promise.resolve() }; } - if ( - (source.kind === 'group' || source.kind === 'group-all') && - this.config.groupPolicy === 'pairing' && - !this.groupGate.isGroupApproved(message.conversationId) - ) { - return; + return this.admitDirectMessage({ kind: 'direct' }, message, key); + } + + private async admitDirectMessage( + source: Extract, + message: DwsImMessage, + key: string, + ): Promise<{ completion: Promise }> { + if (this.cursor.processedMessages.includes(key)) { + this.removePendingMessage(key); + return { completion: Promise.resolve() }; + } + if (!this.hasPendingMessage(key)) { + if (!(await this.rememberPendingMessage(source, message))) { + return { completion: Promise.resolve() }; + } + this.saveCursor(); } - const key = messageKey(message); + return { completion: this.scheduleDirectMessage(source, message, key) }; + } + + private scheduleDirectMessage( + source: Extract, + message: DwsImMessage, + key: string, + ): Promise { + const queued = this.queuedDirectMessages.get(key); + if (queued) return queued; + if (this.cursor.processedMessages.includes(key)) return Promise.resolve(); + const generation = this.lifecycleGeneration; + const task = Promise.resolve().then(async () => { + try { + if (!this.connected || generation !== this.lifecycleGeneration) return; + await this.dispatchImMessage(source, message, key); + } finally { + if (this.queuedDirectMessages.get(key) === task) { + this.queuedDirectMessages.delete(key); + } + } + }); + this.queuedDirectMessages.set(key, task); + void task.catch((error: unknown) => { + this.logImError( + source, + error instanceof Error ? error : new Error(String(error)), + ); + }); + return task; + } + + private async dispatchImMessage( + source: DwsImSource, + message: DwsImMessage, + key: string, + ): Promise { let waitedOnInFlight = false; let inFlightError: unknown; while (true) { @@ -1580,6 +1680,13 @@ export class DwsChannel extends PollingChannelBase { this.processingMessages.set(key, task); try { await task; + if ( + this.cursor.processedMessages.includes(key) && + this.hasPendingMessage(key) + ) { + this.removePendingMessage(key); + this.saveCursor(); + } } finally { if (this.processingMessages.get(key) === task) { this.processingMessages.delete(key); @@ -1651,7 +1758,7 @@ export class DwsChannel extends PollingChannelBase { // Direct-message history may be unavailable, and ambient group // messages have no history fallback. `at` messages need no parking: // the pinned mention checkpoint re-fetches them. - this.rememberPendingMessage(source, message); + await this.rememberPendingMessage(source, message); } // Under budget the throw propagates exactly as before, so redelivery // and concurrent-duplicate retry keep their contracts. Once the budget @@ -1738,30 +1845,41 @@ export class DwsChannel extends PollingChannelBase { ); } - private rememberPendingMessage( + private async rememberPendingMessage( source: PersistedPendingMessage['source'], message: DwsImMessage, - ): void { + ): Promise { const key = messageKey(message); - if (this.hasPendingMessage(key)) return; - const pending = this.cursor.pendingMessages ?? []; - while (pending.length >= MAX_PROCESSED_ITEMS) { - const dropped = pending.shift(); - if (!dropped) break; - this.clearInboundFailure(messageKey(dropped.message)); - process.stderr.write( - `[Channel:${this.name}] dropping the oldest pending DWS message ` + - `because the retry queue reached ${MAX_PROCESSED_ITEMS}.\n`, - ); + if (this.hasPendingMessage(key)) return true; + const generation = this.lifecycleGeneration; + while ( + this.connected && + generation === this.lifecycleGeneration && + (this.cursor.pendingMessages?.length ?? 0) >= MAX_PROCESSED_ITEMS + ) { + await new Promise((resolve) => { + this.pendingMessageCapacityWaiters.add(resolve); + }); } + if (!this.connected || generation !== this.lifecycleGeneration) { + return false; + } + if (this.hasPendingMessage(key)) return true; + const pending = this.cursor.pendingMessages ?? []; pending.push({ source, message }); this.cursor.pendingMessages = pending; + return true; } private removePendingMessage(key: string): void { - this.cursor.pendingMessages = (this.cursor.pendingMessages ?? []).filter( - (pending) => messageKey(pending.message) !== key, + const pending = this.cursor.pendingMessages ?? []; + const remaining = pending.filter( + (item) => messageKey(item.message) !== key, ); + this.cursor.pendingMessages = remaining; + if (remaining.length === pending.length) return; + for (const resolve of this.pendingMessageCapacityWaiters) resolve(); + this.pendingMessageCapacityWaiters.clear(); } private async processDocumentNotification( @@ -1901,8 +2019,31 @@ export class DwsChannel extends PollingChannelBase { for (const pending of [...(this.cursor.pendingMessages ?? [])]) { if (signal.aborted || !this.connected) return; const key = messageKey(pending.message); + if (pending.source.kind === 'direct') { + const alreadyQueued = this.queuedDirectMessages.has(key); + const completion = this.scheduleDirectMessage( + pending.source, + pending.message, + key, + ); + if ( + !alreadyQueued && + parseDocumentMentionNotification(pending.message.content.trim()) !== + undefined + ) { + try { + await completion; + } catch (error) { + if (signal.aborted || !this.connected) return; + process.stderr.write( + `[Channel:${this.name}] pending DWS message remains degraded: ${sanitizeLogText(error instanceof Error ? error.message : String(error), 300)}\n`, + ); + } + } + continue; + } try { - await this.handleImMessage(pending.source, pending.message, true); + await this.receiveImMessage(pending.source, pending.message, true); this.removePendingMessage(key); this.saveCursor(); } catch (error) { diff --git a/packages/channels/dws/src/dws-client.test.ts b/packages/channels/dws/src/dws-client.test.ts index 6ce4840ff86..cbe07873351 100644 --- a/packages/channels/dws/src/dws-client.test.ts +++ b/packages/channels/dws/src/dws-client.test.ts @@ -11,6 +11,8 @@ import { DwsCommandError, parseDwsImEvent, type DwsCommandRunner, + type DwsImDispatch, + type DwsImMessage, } from './dws-client.js'; import type { DwsEventProcessStarter, @@ -450,6 +452,108 @@ describe('DwsClient', () => { }); }); + it('does not make the direct event reader wait for message processing', async () => { + let onLine!: (line: string) => void | Promise; + const eventStarter = vi.fn( + async (_executable, _args, lineHandler) => { + onLine = lineHandler; + return subscription(); + }, + ); + const client = new DwsClient( + { executable: '/opt/dws' }, + vi.fn(), + eventStarter, + ); + let releaseFirst!: () => void; + const firstBlocked = new Promise((resolve) => { + releaseFirst = resolve; + }); + const completed: string[] = []; + const onMessage = vi.fn((message: DwsImMessage): DwsImDispatch => { + const processing = (async () => { + if (message.conversationId === 'conversation-a') await firstBlocked; + completed.push(message.conversationId); + })(); + return { admitted: Promise.resolve(), completed: processing }; + }); + + await client.subscribeToIm({ kind: 'direct' }, onMessage, vi.fn()); + await onLine( + json({ + type: 'user_im_message_receive_o2o_all', + event_id: 'event-a', + message_id: 'message-a', + conversation_id: 'conversation-a', + content: 'first request', + sender_open_dingtalk_id: 'user-a', + sender: 'User A', + }), + ); + await onLine( + json({ + type: 'user_im_message_receive_o2o_all', + event_id: 'event-b', + message_id: 'message-b', + conversation_id: 'conversation-b', + content: 'second request', + sender_open_dingtalk_id: 'user-b', + sender: 'User B', + }), + ); + + expect(onMessage).toHaveBeenCalledTimes(2); + expect(completed).toEqual(['conversation-b']); + releaseFirst(); + await vi.waitFor(() => + expect(completed).toEqual(['conversation-b', 'conversation-a']), + ); + }); + + it('reports a detached direct-message processing failure', async () => { + let onLine!: (line: string) => void | Promise; + const eventStarter = vi.fn( + async (_executable, _args, lineHandler) => { + onLine = lineHandler; + return subscription(); + }, + ); + const client = new DwsClient( + { executable: '/opt/dws' }, + vi.fn(), + eventStarter, + ); + const onError = vi.fn(); + + await client.subscribeToIm( + { kind: 'direct' }, + vi.fn( + (): DwsImDispatch => ({ + admitted: Promise.resolve(), + completed: Promise.reject(new Error('turn failed')), + }), + ), + onError, + ); + await onLine( + json({ + type: 'user_im_message_receive_o2o_all', + event_id: 'event-1', + message_id: 'message-1', + conversation_id: 'conversation-a', + content: 'request', + sender_open_dingtalk_id: 'user-a', + sender: 'User A', + }), + ); + + await vi.waitFor(() => + expect(onError).toHaveBeenCalledWith( + expect.objectContaining({ message: 'turn failed' }), + ), + ); + }); + it('subscribes to all ordinary group messages without a group filter', async () => { const eventStarter = vi.fn(async () => subscription(), diff --git a/packages/channels/dws/src/dws-client.ts b/packages/channels/dws/src/dws-client.ts index 4adb552ae78..58ff26df947 100644 --- a/packages/channels/dws/src/dws-client.ts +++ b/packages/channels/dws/src/dws-client.ts @@ -51,6 +51,13 @@ export interface DwsImMessage { eventTime?: number; } +export interface DwsImDispatch { + admitted: Promise; + completed: Promise; +} + +export type DwsImMessageResult = void | Promise | DwsImDispatch; + export interface DwsTodoTask { taskId: string; title: string; @@ -69,7 +76,7 @@ export interface DwsClientLike { assertAuthenticated(signal?: AbortSignal): Promise; subscribeToIm( source: DwsImSource, - onMessage: (message: DwsImMessage) => void | Promise, + onMessage: (message: DwsImMessage) => DwsImMessageResult, onError: (error: Error) => void, ): Promise; sendImMessage( @@ -750,7 +757,7 @@ export class DwsClient implements DwsClientLike { async subscribeToIm( source: DwsImSource, - onMessage: (message: DwsImMessage) => void | Promise, + onMessage: (message: DwsImMessage) => DwsImMessageResult, onError: (error: Error) => void, ): Promise { const args = [ @@ -767,7 +774,24 @@ export class DwsClient implements DwsClientLike { return this.eventStarter( this.executable, args, - async (line) => onMessage(parseDwsImEvent(line)), + (line) => { + const message = parseDwsImEvent(line); + const result = onMessage(message); + if (!result || !('admitted' in result)) return result; + if (source.kind !== 'direct') return result.completed; + const reportError = (error: unknown): void => { + try { + onError(error instanceof Error ? error : new Error(String(error))); + } catch { + return; + } + }; + void result.admitted.then( + () => result.completed.catch(reportError), + () => undefined, + ); + return result.admitted; + }, onError, ); } diff --git a/packages/channels/dws/src/index.ts b/packages/channels/dws/src/index.ts index 6980e58963a..e1c3c3e7193 100644 --- a/packages/channels/dws/src/index.ts +++ b/packages/channels/dws/src/index.ts @@ -14,7 +14,9 @@ export type { DwsClientLike, DwsClientOptions, DwsIdentity, + DwsImDispatch, DwsImMessage, + DwsImMessageResult, DwsImSource, DwsImTarget, DwsMessageHistoryPage, From 36fc591078403b33770a11bd821509bc494f2a24 Mon Sep 17 00:00:00 2001 From: qqqys <266654365+qqqys@users.noreply.github.com> Date: Tue, 1 Sep 2026 17:30:27 +0800 Subject: [PATCH 2/5] fix(dws): harden concurrent direct dispatch --- packages/channels/dws/src/dws-channel.test.ts | 171 +++++++++++++++--- packages/channels/dws/src/dws-channel.ts | 86 ++++----- packages/channels/dws/src/dws-client.test.ts | 54 ++++++ packages/channels/dws/src/dws-client.ts | 9 +- 4 files changed, 241 insertions(+), 79 deletions(-) diff --git a/packages/channels/dws/src/dws-channel.test.ts b/packages/channels/dws/src/dws-channel.test.ts index dac7ebcb77b..640ead322b2 100644 --- a/packages/channels/dws/src/dws-channel.test.ts +++ b/packages/channels/dws/src/dws-channel.test.ts @@ -240,8 +240,10 @@ class FakeDwsClient implements DwsClientLike { const stream = this.streams[sourceIndex]; if (!stream) throw new Error(`Missing fake stream ${sourceIndex}.`); const result = stream.onMessage(event); - if (result && 'completed' in result) await result.completed; - else await result; + if (result && 'completed' in result) { + await result.admitted; + await result.completed; + } else await result; } } @@ -249,6 +251,7 @@ class TestableDwsChannel extends DwsChannel { inbound: Envelope[] = []; inboundError?: Error; inboundHandler?: (envelope: Envelope) => Promise; + nextCursorSaveError?: Error; responseMessageId?: string; responseSenderId?: string; responseThreadId?: string; @@ -259,6 +262,15 @@ class TestableDwsChannel extends DwsChannel { return 0; } + protected override saveCursor(): void { + if (this.nextCursorSaveError) { + const error = this.nextCursorSaveError; + this.nextCursorSaveError = undefined; + throw error; + } + super.saveCursor(); + } + inboundAttempts = 0; override async handleInbound(envelope: Envelope): Promise { @@ -348,12 +360,17 @@ class TestableDwsChannel extends DwsChannel { releasePendingMessage(conversationId: string, messageId: string): void { const removePendingMessage = ( - this as unknown as { removePendingMessage(key: string): void } + this as unknown as { removePendingMessage(key: string): boolean } ).removePendingMessage.bind(this); removePendingMessage(`${conversationId}\0${messageId}`); this.saveCursor(); } + markPendingMessageProcessed(conversationId: string, messageId: string): void { + this.cursor.processedMessages.push(`${conversationId}\0${messageId}`); + this.saveCursor(); + } + resolveSession(): Promise { return this.router.resolve(this.name, 'alice', 'doc-1', 'comment-1'); } @@ -399,6 +416,14 @@ class PolicyDwsChannel extends DwsChannel { return this.cursor.pendingDocumentNotifications ?? []; } + queuedDirectMessageCount(): number { + return ( + this as unknown as { + queuedDirectMessages: Map>; + } + ).queuedDirectMessages.size; + } + documentSetSize(): number { return (this as unknown as { documentSet: Set }).documentSet.size; } @@ -1476,6 +1501,38 @@ describe('DwsChannel', () => { expect(channel.pendingMessageIds()).toHaveLength(4_999); }); + it('does not acknowledge admission when cursor persistence fails', async () => { + const client = new FakeDwsClient(); + const channel = await readyChannel(client); + const event = message( + 'user_im_message_receive_o2o_all', + 'retry-after-save-failure', + 'request', + ); + channel.nextCursorSaveError = new Error('disk unavailable'); + + await expect(client.emit(1, event)).rejects.toThrow('disk unavailable'); + expect(channel.pendingMessageIds()).toEqual([]); + expect(channel.inbound).toEqual([]); + + await client.emit(1, event); + expect(channel.inbound.map(({ messageId }) => messageId)).toEqual([ + 'retry-after-save-failure', + ]); + }); + + it('cleans up a persisted message that is already processed', async () => { + const client = new FakeDwsClient(); + const channel = await readyChannel(client); + channel.seedPendingMessages(1); + channel.markPendingMessageProcessed('conversation-capacity', 'parked-0'); + + await channel.poll(); + + expect(channel.pendingMessageIds()).toEqual([]); + expect(channel.inbound).toEqual([]); + }); + it('turns a document mention notification into a document task and replies to its comment', async () => { const client = new FakeDwsClient(); const channel = await readyChannel(client); @@ -2022,6 +2079,55 @@ describe('DwsChannel', () => { await vi.waitFor(() => expect(channel.inbound).toHaveLength(2)); }); + it('does not let a document notification block another history conversation', async () => { + const client = new FakeDwsClient(); + const channel = await readyChannel(client); + let releaseDocument!: () => void; + const documentBlocked = new Promise((resolve) => { + releaseDocument = resolve; + }); + let releaseOrdinary!: () => void; + const ordinaryBlocked = new Promise((resolve) => { + releaseOrdinary = resolve; + }); + const started: string[] = []; + channel.inboundHandler = async (envelope) => { + started.push(envelope.messageId); + if (envelope.messageId === 'history-document') await documentBlocked; + if (envelope.messageId === 'history-ordinary') await ordinaryBlocked; + channel.inbound.push(envelope); + }; + const eventTime = Date.now(); + client.directMessages = [ + message( + 'user_im_message_receive_o2o_all', + 'history-document', + documentMentionCard('doc-history-blocked', 'comment-history-blocked'), + { conversationId: 'conversation-document', eventTime }, + ), + message( + 'user_im_message_receive_o2o_all', + 'history-ordinary', + 'second request', + { conversationId: 'conversation-ordinary', eventTime }, + ), + ]; + + await channel.poll(); + + await vi.waitFor(() => + expect(started).toEqual(['history-document', 'history-ordinary']), + ); + expect(channel.pendingMessageIds()).toEqual([ + 'history-document', + 'history-ordinary', + ]); + releaseDocument(); + releaseOrdinary(); + await vi.waitFor(() => expect(channel.inbound).toHaveLength(2)); + expect(channel.pendingMessageIds()).toEqual([]); + }); + // R1-7: every history window re-opens at `watermark - 5s`, so every // live-dispatched direct message is re-fetched by a later poll. The // processed-key guard is the only thing standing between that refetch and @@ -2043,10 +2149,11 @@ describe('DwsChannel', () => { }); // R2-1: the history loop dispatches every DM-history message, and - // `handleImMessage`'s self-message check is the only filter keeping the - // bot's own replies — now ordinary sent messages that reappear in every - // overlap window — out of the agent. If that check were ever conditioned - // on `!fromHistory`, every poll would re-dispatch them as fresh turns. + // The self-message check in `admitReceivedDirectMessage` is the only filter + // keeping the bot's own replies — now ordinary sent messages that reappear + // in every overlap window — out of the agent. If that check were ever + // conditioned on `!fromHistory`, every poll would re-dispatch them as fresh + // turns. it('does not dispatch self-sent messages recovered from direct-message history', async () => { const client = new FakeDwsClient(); const channel = await readyChannel(client); @@ -2064,9 +2171,9 @@ describe('DwsChannel', () => { }); // R1-1 (fix-induced): without the loop-level processed-key skip, every - // re-fetched self-message re-enters `handleImMessage`, whose self branch - // persists the whole cursor before the processed-key early return — one - // blocking mkdir/write/rename per own reply per poll on top of the + // re-fetched self-message re-enters `admitReceivedDirectMessage`, whose self + // branch persists the whole cursor before the processed-key early return — + // one blocking mkdir/write/rename per own reply per poll on top of the // end-of-poll persist. it('saves the cursor once per poll for own replies re-fetched in the overlap window', async () => { const client = new FakeDwsClient(); @@ -2130,12 +2237,10 @@ describe('DwsChannel', () => { } }); - // R2-2 discriminator: a document notification whose turn fails is NOT - // parked, so the mid-window catch must rethrow it — the pinned watermark - // is its only retry path. Swallowing it would advance the watermark, the - // notification would fall out of the overlap window, and its remaining - // budget would never run. - it('keeps spending the retry budget of an unparked document notification', async () => { + // R2-2 discriminator: a failed document notification is durably admitted + // before its turn starts. The detached turn must leave that pending entry + // available for later polls until the shared retry budget is exhausted. + it('keeps spending the retry budget of a parked document notification', async () => { vi.useFakeTimers(); try { const client = new FakeDwsClient(); @@ -3001,10 +3106,12 @@ describe('DwsChannel', () => { await channel.poll(); - expect(channel.pendingDocumentNotifications()).toEqual([ - expect.objectContaining({ senderId: 'open-alice' }), - expect.objectContaining({ senderId: 'open-bob' }), - ]); + await vi.waitFor(() => + expect(channel.pendingDocumentNotifications()).toEqual([ + expect.objectContaining({ senderId: 'open-alice' }), + expect.objectContaining({ senderId: 'open-bob' }), + ]), + ); const bobPairingText = client.sendImMessage.mock.calls.find( ([target]) => target.kind === 'direct' && target.openDingTalkId === 'open-bob', @@ -3017,8 +3124,10 @@ describe('DwsChannel', () => { await channel.poll(); await channel.poll(); - expect(bridge.prompt).toHaveBeenCalledOnce(); - expect(channel.pendingDocumentNotifications()).toEqual([]); + await vi.waitFor(() => expect(bridge.prompt).toHaveBeenCalledOnce()); + await vi.waitFor(() => + expect(channel.pendingDocumentNotifications()).toEqual([]), + ); }); it('drops profile-scoped document work and IM targets on profile switch', async () => { @@ -4730,8 +4839,8 @@ describe('DwsChannel', () => { releasePairing(); await Promise.all([denied, catchUpPoll]); - expect(bridge.prompt).not.toHaveBeenCalled(); - expect(channel.pendingDocumentNotifications()).toContainEqual( + await vi.waitFor(() => expect(bridge.prompt).toHaveBeenCalledOnce()); + expect(channel.pendingDocumentNotifications()).not.toContainEqual( expect.objectContaining({ messageId: 'allowed-catch-up' }), ); expect(channel.notificationWatermark()).toBeGreaterThan( @@ -5038,7 +5147,7 @@ describe('DwsChannel', () => { // R3-1: history dispatch skips messages that are ALREADY parked, but a // direct message whose live turn is still in flight passes the skip and - // blocks in `handleImMessage`'s in-flight wait. When the live turn then + // blocks in `dispatchImMessage`'s in-flight wait. When the live turn then // fails it parks the message and spends attempt 1 — parked ≠ processed, so // the waiting history dispatch must not start a second turn in the same // poll and spend attempt 2. @@ -5262,6 +5371,12 @@ describe('DwsChannel', () => { for (let round = 0; round < 6; round += 1) { await channel.poll(); + await vi.waitFor(() => + expect(bridge.prompt).toHaveBeenCalledTimes(Math.min(round + 1, 5)), + ); + await vi.waitFor(() => + expect(channel.queuedDirectMessageCount()).toBe(0), + ); } expect(bridge.prompt).toHaveBeenCalledTimes(5); @@ -5282,8 +5397,10 @@ describe('DwsChannel', () => { await channel.poll(); - expect(bridge.prompt).toHaveBeenCalledTimes(6); - expect(channel.pendingDocumentNotifications()).toEqual([]); + await vi.waitFor(() => expect(bridge.prompt).toHaveBeenCalledTimes(6)); + await vi.waitFor(() => + expect(channel.pendingDocumentNotifications()).toEqual([]), + ); }); // R4-1: `pollTodos` remembers a fingerprint only on success, so a todo whose diff --git a/packages/channels/dws/src/dws-channel.ts b/packages/channels/dws/src/dws-channel.ts index ca282d01017..1b41b0d45fb 100644 --- a/packages/channels/dws/src/dws-channel.ts +++ b/packages/channels/dws/src/dws-channel.ts @@ -1177,25 +1177,12 @@ export class DwsChannel extends PollingChannelBase { // `replayPendingMessages`; dispatching it here too would spend the // shared retry budget twice per poll. if (this.hasPendingMessage(key)) continue; - try { - const dispatch = this.receiveDirectMessage(message, true); - await (parseDocumentMentionNotification(message.content.trim()) - ? dispatch.completed - : dispatch.admitted); - } catch (error) { - // A failed plain direct message was just parked for replay, so the - // page can keep moving. An unparked failure — a document - // notification's turn — must still abort the window: the pinned - // watermark is what re-fetches it until its budget is spent. - if (!this.hasPendingMessage(key)) throw error; - process.stderr.write( - `[Channel:${this.name}] direct-message dispatch failed mid-window; the message is parked for retry: ${sanitizeLogText(error instanceof Error ? error.message : String(error), 300)}\n`, - ); - } + await this.receiveDirectMessage(message, true).admitted; } if (this.notificationWatermarkPulledBack) { // R4-4: a stale direct message replayed while this window's - // fetch was in flight, and `handleImMessage` pulled the watermark back + // fetch was in flight, and `admitReceivedDirectMessage` pulled the + // watermark back // to cover it. That replay was left UNMARKED on purpose for history // polling, so finishing this window normally would undo the rescue: // `checkpoint.endTime` is always past the replay's `eventTime`, and @@ -1612,14 +1599,13 @@ export class DwsChannel extends PollingChannelBase { key: string, ): Promise<{ completion: Promise }> { if (this.cursor.processedMessages.includes(key)) { - this.removePendingMessage(key); + if (this.removePendingMessage(key)) this.saveCursor(); return { completion: Promise.resolve() }; } if (!this.hasPendingMessage(key)) { if (!(await this.rememberPendingMessage(source, message))) { return { completion: Promise.resolve() }; } - this.saveCursor(); } return { completion: this.scheduleDirectMessage(source, message, key) }; } @@ -1631,7 +1617,10 @@ export class DwsChannel extends PollingChannelBase { ): Promise { const queued = this.queuedDirectMessages.get(key); if (queued) return queued; - if (this.cursor.processedMessages.includes(key)) return Promise.resolve(); + if (this.cursor.processedMessages.includes(key)) { + if (this.removePendingMessage(key)) this.saveCursor(); + return Promise.resolve(); + } const generation = this.lifecycleGeneration; const task = Promise.resolve().then(async () => { try { @@ -1868,18 +1857,25 @@ export class DwsChannel extends PollingChannelBase { const pending = this.cursor.pendingMessages ?? []; pending.push({ source, message }); this.cursor.pendingMessages = pending; + try { + this.saveCursor(); + } catch (error) { + this.removePendingMessage(key); + throw error; + } return true; } - private removePendingMessage(key: string): void { + private removePendingMessage(key: string): boolean { const pending = this.cursor.pendingMessages ?? []; const remaining = pending.filter( (item) => messageKey(item.message) !== key, ); this.cursor.pendingMessages = remaining; - if (remaining.length === pending.length) return; + if (remaining.length === pending.length) return false; for (const resolve of this.pendingMessageCapacityWaiters) resolve(); this.pendingMessageCapacityWaiters.clear(); + return true; } private async processDocumentNotification( @@ -1902,18 +1898,24 @@ export class DwsChannel extends PollingChannelBase { // only re-drives a parked entry whose own `senderId` passes the gate // (the denied one never will), and this key is skipped by every later // history poll, so the allowed user's request went unanswered forever. - // Mark only when the comment is genuinely done, or when this caller is - // no more entitled to it than the sender already parked. + // Mark only when the comment is genuinely done, or when this sender's + // own request is already parked. Other senders must take the slot next. if ( this.cursor.processedMessages.includes(notificationKey) || - (this.hasPendingDocumentNotification(notificationKey) && + (this.hasPendingDocumentNotification( + notificationKey, + message.senderId, + ) && !this.gate.isAllowed(message.senderId)) ) { this.markProcessedMessage(key); - } else { - this.rememberPendingDocumentNotification(message, notification); + this.saveCursor(); + return; } - this.saveCursor(); + if (this.processingMessages.get(notificationKey) === inFlight) { + this.processingMessages.delete(notificationKey); + } + await this.processDocumentNotification(message, key, notification); return; } const task = (async () => { @@ -2020,26 +2022,7 @@ export class DwsChannel extends PollingChannelBase { if (signal.aborted || !this.connected) return; const key = messageKey(pending.message); if (pending.source.kind === 'direct') { - const alreadyQueued = this.queuedDirectMessages.has(key); - const completion = this.scheduleDirectMessage( - pending.source, - pending.message, - key, - ); - if ( - !alreadyQueued && - parseDocumentMentionNotification(pending.message.content.trim()) !== - undefined - ) { - try { - await completion; - } catch (error) { - if (signal.aborted || !this.connected) return; - process.stderr.write( - `[Channel:${this.name}] pending DWS message remains degraded: ${sanitizeLogText(error instanceof Error ? error.message : String(error), 300)}\n`, - ); - } - } + this.scheduleDirectMessage(pending.source, pending.message, key); continue; } try { @@ -2102,9 +2085,14 @@ export class DwsChannel extends PollingChannelBase { ); } - private hasPendingDocumentNotification(notificationKey: string): boolean { + private hasPendingDocumentNotification( + notificationKey: string, + senderId?: string, + ): boolean { return (this.cursor.pendingDocumentNotifications ?? []).some( - (pending) => documentNotificationKey(pending) === notificationKey, + (pending) => + documentNotificationKey(pending) === notificationKey && + (senderId === undefined || pending.senderId === senderId), ); } diff --git a/packages/channels/dws/src/dws-client.test.ts b/packages/channels/dws/src/dws-client.test.ts index cbe07873351..6971306398a 100644 --- a/packages/channels/dws/src/dws-client.test.ts +++ b/packages/channels/dws/src/dws-client.test.ts @@ -554,6 +554,60 @@ describe('DwsClient', () => { ); }); + it('observes completion failure while direct admission is pending', async () => { + let onLine!: (line: string) => void | Promise; + const eventStarter = vi.fn( + async (_executable, _args, lineHandler) => { + onLine = lineHandler; + return subscription(); + }, + ); + const client = new DwsClient( + { executable: '/opt/dws' }, + vi.fn(), + eventStarter, + ); + let releaseAdmission!: () => void; + const admitted = new Promise((resolve) => { + releaseAdmission = resolve; + }); + let rejectCompletion!: (error: Error) => void; + const completed = new Promise((_resolve, reject) => { + rejectCompletion = reject; + }); + const onError = vi.fn(); + + await client.subscribeToIm( + { kind: 'direct' }, + vi.fn((): DwsImDispatch => ({ admitted, completed })), + onError, + ); + const reading = Promise.resolve( + onLine( + json({ + type: 'user_im_message_receive_o2o_all', + event_id: 'event-1', + message_id: 'message-1', + conversation_id: 'conversation-a', + content: 'request', + sender_open_dingtalk_id: 'user-a', + sender: 'User A', + }), + ), + ); + rejectCompletion(new Error('turn failed')); + await new Promise((resolve) => setTimeout(resolve, 0)); + expect(onError).not.toHaveBeenCalled(); + + releaseAdmission(); + await reading; + await vi.waitFor(() => + expect(onError).toHaveBeenCalledWith( + expect.objectContaining({ message: 'turn failed' }), + ), + ); + }); + it('subscribes to all ordinary group messages without a group filter', async () => { const eventStarter = vi.fn(async () => subscription(), diff --git a/packages/channels/dws/src/dws-client.ts b/packages/channels/dws/src/dws-client.ts index 58ff26df947..1ef21c53d00 100644 --- a/packages/channels/dws/src/dws-client.ts +++ b/packages/channels/dws/src/dws-client.ts @@ -786,10 +786,13 @@ export class DwsClient implements DwsClientLike { return; } }; - void result.admitted.then( - () => result.completed.catch(reportError), - () => undefined, + const admissionSucceeded = result.admitted.then( + () => true, + () => false, ); + void result.completed.catch(async (error: unknown) => { + if (await admissionSucceeded) reportError(error); + }); return result.admitted; }, onError, From cf2084fe69239d57d8a367ff9d1f6f746b5c6ad4 Mon Sep 17 00:00:00 2001 From: qqqys <266654365+qqqys@users.noreply.github.com> Date: Tue, 1 Sep 2026 23:53:12 +0800 Subject: [PATCH 3/5] fix(dws): address concurrent dispatch review --- packages/channels/dws/src/dws-channel.test.ts | 412 +++++++++++++++++- packages/channels/dws/src/dws-channel.ts | 157 +++++-- packages/channels/dws/src/dws-client.test.ts | 43 ++ 3 files changed, 573 insertions(+), 39 deletions(-) diff --git a/packages/channels/dws/src/dws-channel.test.ts b/packages/channels/dws/src/dws-channel.test.ts index 640ead322b2..15fde392013 100644 --- a/packages/channels/dws/src/dws-channel.test.ts +++ b/packages/channels/dws/src/dws-channel.test.ts @@ -12,6 +12,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { PairingStore, type ChannelAgentBridge, + type ChannelBaseOptions, type ChannelConfig, type Envelope, } from '@qwen-code/channel-base'; @@ -63,6 +64,23 @@ function makeBridge(): ChannelAgentBridge { } as unknown as ChannelAgentBridge; } +function makeChannelMemory(): NonNullable { + return { + readChannelMemory: vi.fn().mockResolvedValue(''), + listChannelMemoryEntries: vi.fn().mockResolvedValue([]), + addChannelMemoryEntries: vi.fn().mockResolvedValue({ + changed: false, + added: [], + duplicateIds: [], + }), + updateChannelMemoryEntry: vi.fn().mockResolvedValue({ changed: false }), + removeChannelMemoryEntries: vi + .fn() + .mockResolvedValue({ changed: false, removed: [] }), + clearChannelMemory: vi.fn().mockResolvedValue({ changed: false }), + }; +} + function message( type: DwsImMessage['type'], messageId: string, @@ -342,7 +360,7 @@ class TestableDwsChannel extends DwsChannel { ); } - seedPendingMessages(count: number): void { + seedPendingMessages(count: number, separateConversations = false): void { this.cursor.pendingMessages = Array.from( { length: count }, (_unused, index) => ({ @@ -351,7 +369,11 @@ class TestableDwsChannel extends DwsChannel { 'user_im_message_receive_o2o_all', `parked-${index}`, `request ${index}`, - { conversationId: 'conversation-capacity' }, + { + conversationId: separateConversations + ? `conversation-capacity-${index}` + : 'conversation-capacity', + }, ), }), ); @@ -399,6 +421,30 @@ class TestableDwsChannel extends DwsChannel { inboundFailures(): unknown[] { return this.cursor.inboundFailures ?? []; } + + pendingMessageCapacityWaiterCount(): number { + return ( + this as unknown as { + pendingMessageCapacityWaiters: Set<() => void>; + } + ).pendingMessageCapacityWaiters.size; + } + + queuedDirectMessage(key: string): Promise | undefined { + return ( + this as unknown as { + queuedDirectMessages: Map>; + } + ).queuedDirectMessages.get(key); + } + + replaceQueuedDirectMessage(key: string, task: Promise): void { + ( + this as unknown as { + queuedDirectMessages: Map>; + } + ).queuedDirectMessages.set(key, task); + } } class PolicyDwsChannel extends DwsChannel { @@ -500,9 +546,10 @@ async function readyPolicyChannel( client: FakeDwsClient, config = makeConfig(), name = 'policy-dws', + options?: ChannelBaseOptions, ): Promise<{ channel: PolicyDwsChannel; bridge: ChannelAgentBridge }> { const bridge = makeBridge(); - const channel = new PolicyDwsChannel(name, config, bridge, undefined, client); + const channel = new PolicyDwsChannel(name, config, bridge, options, client); channels.push(channel); await channel.connect(); return { channel, bridge }; @@ -1475,6 +1522,210 @@ describe('DwsChannel', () => { expect(bridge.prompt).toHaveBeenCalledTimes(2); }); + it('preserves followup order while the first message is classified', async () => { + const client = new FakeDwsClient(); + let finishClassification!: (result: { + intent: 'none'; + confidence: number; + }) => void; + const classification = new Promise<{ + intent: 'none'; + confidence: number; + }>((resolve) => { + finishClassification = resolve; + }); + const memoryIntentClassifier = { + classifyChannelMemoryIntent: vi + .fn() + .mockReturnValueOnce(classification) + .mockResolvedValue({ intent: 'none', confidence: 0.9 }), + }; + const { bridge } = await readyPolicyChannel( + client, + makeConfig({ dispatchMode: 'followup' }), + 'classified-order-dws', + { + channelMemory: makeChannelMemory(), + memoryIntentClassifier, + }, + ); + + const firstDelivery = client.emit( + 1, + message( + 'user_im_message_receive_o2o_all', + 'classified-first', + 'remember this please', + { conversationId: 'conversation-a' }, + ), + ); + await vi.waitFor(() => + expect( + memoryIntentClassifier.classifyChannelMemoryIntent, + ).toHaveBeenCalledOnce(), + ); + const secondDelivery = client.emit( + 1, + message( + 'user_im_message_receive_o2o_all', + 'classified-second', + 'what time is it', + { conversationId: 'conversation-a' }, + ), + ); + + await new Promise((resolve) => setTimeout(resolve, 0)); + expect(bridge.prompt).not.toHaveBeenCalled(); + finishClassification({ intent: 'none', confidence: 0.9 }); + await Promise.all([firstDelivery, secondDelivery]); + expect( + (bridge.prompt as ReturnType).mock.calls.map( + (call) => call[1] as string, + ), + ).toEqual([ + expect.stringContaining('remember this please'), + expect.stringContaining('what time is it'), + ]); + }); + + it('caps concurrent direct-message replay dispatches', async () => { + const client = new FakeDwsClient(); + const channel = await readyChannel(client); + channel.seedPendingMessages(25, true); + let release!: () => void; + const blocked = new Promise((resolve) => { + release = resolve; + }); + channel.inboundHandler = async (envelope) => { + await blocked; + channel.inbound.push(envelope); + }; + + await channel.poll(); + await vi.waitFor(() => expect(channel.inboundAttempts).toBe(16)); + expect(channel.inboundAttempts).toBeLessThan(25); + + release(); + await vi.waitFor(() => expect(channel.pendingMessageIds()).toHaveLength(9)); + }); + + it('clears queued direct messages when disconnected', async () => { + const client = new FakeDwsClient(); + const channel = await readyChannel(client); + let release!: () => void; + const blocked = new Promise((resolve) => { + release = resolve; + }); + channel.inboundHandler = async () => blocked; + const event = message( + 'user_im_message_receive_o2o_all', + 'disconnect-queued', + 'request', + ); + const key = `${event.conversationId}\0${event.messageId}`; + + const delivery = client.emit(1, event); + await vi.waitFor(() => + expect(channel.queuedDirectMessage(key)).toBeDefined(), + ); + channel.disconnect(); + + expect(channel.queuedDirectMessage(key)).toBeUndefined(); + release(); + await delivery; + }); + + it('releases capacity-blocked admission when disconnected', async () => { + const client = new FakeDwsClient(); + const channel = await readyChannel(client); + channel.seedPendingMessages(5_000); + + const delivery = client.emit( + 1, + message( + 'user_im_message_receive_o2o_all', + 'disconnect-capacity', + 'request', + ), + ); + await vi.waitFor(() => + expect(channel.pendingMessageCapacityWaiterCount()).toBe(1), + ); + channel.disconnect(); + + await delivery; + expect(channel.pendingMessageCapacityWaiterCount()).toBe(0); + }); + + it('does not let an older task delete a replacement queue entry', async () => { + const client = new FakeDwsClient(); + const channel = await readyChannel(client); + let release!: () => void; + const blocked = new Promise((resolve) => { + release = resolve; + }); + channel.inboundHandler = async () => blocked; + const event = message( + 'user_im_message_receive_o2o_all', + 'replaced-queue-entry', + 'request', + ); + const key = `${event.conversationId}\0${event.messageId}`; + const delivery = client.emit(1, event); + await vi.waitFor(() => + expect(channel.queuedDirectMessage(key)).toBeDefined(), + ); + const replacement = new Promise(() => undefined); + channel.replaceQueuedDirectMessage(key, replacement); + + release(); + await delivery; + + expect(channel.queuedDirectMessage(key)).toBe(replacement); + }); + + it('reports one stream error for a failed live direct turn', async () => { + const stderr = vi.spyOn(process.stderr, 'write').mockReturnValue(true); + try { + const client = new FakeDwsClient(); + const channel = await readyChannel(client); + channel.inboundError = new Error('agent unavailable'); + const stream = client.streams[1]!; + const result = stream.onMessage( + message( + 'user_im_message_receive_o2o_all', + 'single-live-error', + 'request', + ), + ); + if (!result || !('completed' in result)) { + throw new Error('Expected a detached direct-message dispatch.'); + } + const admissionSucceeded = result.admitted.then( + () => true, + () => false, + ); + void result.completed.catch(async (error: unknown) => { + if (await admissionSucceeded) { + stream.onError( + error instanceof Error ? error : new Error(String(error)), + ); + } + }); + + await result.admitted; + await expect(result.completed).rejects.toThrow('agent unavailable'); + await vi.waitFor(() => { + const errors = stderr.mock.calls + .map((call) => String(call[0])) + .filter((line) => line.includes('DWS direct messages stream error')); + expect(errors).toHaveLength(1); + }); + } finally { + stderr.mockRestore(); + } + }); + it('applies backpressure instead of evicting an admitted message', async () => { const client = new FakeDwsClient(); const channel = await readyChannel(client); @@ -1501,6 +1752,68 @@ describe('DwsChannel', () => { expect(channel.pendingMessageIds()).toHaveLength(4_999); }); + it('does not advance direct history past admission released by disconnect', async () => { + const client = new FakeDwsClient(); + const channel = await readyChannel(client); + const initialWatermark = channel.notificationWatermark()!; + channel.seedPendingMessages(5_000, true); + let release!: () => void; + const blocked = new Promise((resolve) => { + release = resolve; + }); + channel.inboundHandler = async (envelope) => { + await blocked; + channel.inbound.push(envelope); + }; + const victim = message( + 'user_im_message_receive_o2o_all', + 'history-capacity-victim', + 'request', + { + conversationId: 'history-victim', + eventTime: initialWatermark - 1_000, + }, + ); + client.directMessages = [victim]; + const now = vi + .spyOn(Date, 'now') + .mockReturnValue(initialWatermark + 10_000); + try { + const poll = channel.poll(); + await vi.waitFor(() => + expect(channel.pendingMessageCapacityWaiterCount()).toBe(1), + ); + channel.disconnect(); + await poll; + + expect(channel.notificationWatermark()).toBe(initialWatermark); + + release(); + await vi.waitFor(() => + expect(channel.pendingMessageIds()).toHaveLength(4_984), + ); + channel.inboundHandler = async (envelope) => { + channel.inbound.push(envelope); + }; + client.listDirectMessages.mockClear(); + await channel.connect(); + await channel.poll(); + await vi.waitFor(() => + expect( + channel.inbound.some( + ({ messageId }) => messageId === 'history-capacity-victim', + ), + ).toBe(true), + ); + expect(client.listDirectMessages.mock.calls[0]![0]).toBeLessThanOrEqual( + victim.eventTime!, + ); + } finally { + release(); + now.mockRestore(); + } + }); + it('does not acknowledge admission when cursor persistence fails', async () => { const client = new FakeDwsClient(); const channel = await readyChannel(client); @@ -4984,6 +5297,99 @@ describe('DwsChannel', () => { expect(channel.mentionWatermark()).toBeGreaterThan(0); }); + it('retains a failed ambient message when its parking save fails', async () => { + const client = new FakeDwsClient(); + const channel = await readyChannel( + client, + makeConfig({ groups: { '*': { requireMention: false } } }), + ); + channel.inboundError = new Error('agent unavailable'); + channel.nextCursorSaveError = new Error('disk unavailable'); + const event = message( + 'user_im_message_receive_group_all', + 'ambient-save-failure', + 'please retry this group request', + { conversationId: 'cid-group' }, + ); + + await expect(client.emit(1, event)).rejects.toThrow('agent unavailable'); + expect(channel.pendingMessageIds()).toEqual(['ambient-save-failure']); + expect(channel.inboundFailures()).toEqual([ + expect.objectContaining({ attempts: 1 }), + ]); + + channel.inboundError = undefined; + await channel.poll(); + expect(channel.inbound).toEqual([ + expect.objectContaining({ messageId: 'ambient-save-failure' }), + ]); + }); + + it('keeps failed ambient parking behind the pending-message cap', async () => { + const client = new FakeDwsClient(); + const channel = await readyChannel( + client, + makeConfig({ groups: { '*': { requireMention: false } } }), + ); + channel.seedPendingMessages(5_000); + channel.inboundError = new Error('agent unavailable'); + const event = message( + 'user_im_message_receive_group_all', + 'ambient-at-capacity', + 'please retry this group request', + { conversationId: 'cid-group' }, + ); + + const delivery = client.emit(1, event); + const failedDelivery = + expect(delivery).rejects.toThrow('agent unavailable'); + await vi.waitFor(() => + expect(channel.pendingMessageCapacityWaiterCount()).toBe(1), + ); + expect(channel.pendingMessageIds()).toHaveLength(5_000); + expect(channel.pendingMessageIds()).not.toContain('ambient-at-capacity'); + + channel.releasePendingMessage('conversation-capacity', 'parked-0'); + await failedDelivery; + expect(channel.pendingMessageIds()).toHaveLength(5_000); + expect(channel.pendingMessageIds()).toContain('ambient-at-capacity'); + }); + + it('retains a failed ambient message across a disconnect', async () => { + const client = new FakeDwsClient(); + const channel = await readyChannel( + client, + makeConfig({ groups: { '*': { requireMention: false } } }), + ); + let rejectTurn!: (error: Error) => void; + const turn = new Promise((_resolve, reject) => { + rejectTurn = reject; + }); + channel.inboundHandler = async () => turn; + const event = message( + 'user_im_message_receive_group_all', + 'ambient-disconnect', + 'please retry this group request', + { conversationId: 'cid-group' }, + ); + + const delivery = client.emit(1, event); + await vi.waitFor(() => expect(channel.inboundAttempts).toBe(1)); + channel.disconnect(); + rejectTurn(new Error('agent unavailable')); + await expect(delivery).rejects.toThrow('agent unavailable'); + expect(channel.pendingMessageIds()).toEqual(['ambient-disconnect']); + + channel.inboundHandler = async (envelope) => { + channel.inbound.push(envelope); + }; + await channel.connect(); + await channel.poll(); + expect(channel.inbound).toEqual([ + expect.objectContaining({ messageId: 'ambient-disconnect' }), + ]); + }); + it('replays a failed ambient group message after restart', async () => { const config = makeConfig({ groups: { '*': { requireMention: false } } }); const name = 'pending-ambient-group-dws'; diff --git a/packages/channels/dws/src/dws-channel.ts b/packages/channels/dws/src/dws-channel.ts index 1b41b0d45fb..977846f45a8 100644 --- a/packages/channels/dws/src/dws-channel.ts +++ b/packages/channels/dws/src/dws-channel.ts @@ -38,6 +38,7 @@ const MAX_DOCUMENT_CONTEXT_CHARS = 12_000; const MAX_TODO_CONTEXT_CHARS = 12_000; const MAX_COMMENT_CHARS = 4_000; const MAX_PROCESSED_ITEMS = 5_000; +const MAX_DIRECT_REPLAY_DISPATCHES = 16; /** * How many times one inbound message may fail its turn before it is dropped. * @@ -520,6 +521,7 @@ export class DwsChannel extends PollingChannelBase { private readonly notifiedSenderPairingNotifications = new Set(); private readonly processingMessages = new Map>(); private readonly queuedDirectMessages = new Map>(); + private readonly directConversationTails = new Map>(); private readonly pendingMessageCapacityWaiters = new Set<() => void>(); private pollAbortController = new AbortController(); private lifecycleGeneration = 0; @@ -829,6 +831,7 @@ export class DwsChannel extends PollingChannelBase { } this.sessionReactionKeys.clear(); this.queuedDirectMessages.clear(); + this.directConversationTails.clear(); for (const resolve of this.pendingMessageCapacityWaiters) resolve(); this.pendingMessageCapacityWaiters.clear(); this.stopPollLoop(); @@ -1179,6 +1182,7 @@ export class DwsChannel extends PollingChannelBase { if (this.hasPendingMessage(key)) continue; await this.receiveDirectMessage(message, true).admitted; } + if (signal.aborted || !this.connected) return; if (this.notificationWatermarkPulledBack) { // R4-4: a stale direct message replayed while this window's // fetch was in flight, and `admitReceivedDirectMessage` pulled the @@ -1492,24 +1496,15 @@ export class DwsChannel extends PollingChannelBase { } private async receiveImMessage( - source: DwsImSource, + source: Exclude, message: DwsImMessage, fromHistory = false, ): Promise { - if (source.kind === 'direct') { - await this.receiveDirectMessage(message, fromHistory).completed; - return; - } if (!this.connected) return; - if (this.isSelfMessage(message)) { - this.markProcessedMessage(messageKey(message)); - this.saveCursor(); - return; - } + if (this.markSelfMessageProcessed(message)) return; if ( - !fromHistory && - message.eventTime !== undefined && - message.eventTime < this.connectionStartedAt - 5_000 + this.isStaleLiveMessage(message, fromHistory) && + message.eventTime !== undefined ) { this.markProcessedMessage(messageKey(message)); this.saveCursor(); @@ -1536,8 +1531,13 @@ export class DwsChannel extends PollingChannelBase { private receiveDirectMessage( message: DwsImMessage, fromHistory = false, + reportFailure = fromHistory, ): DwsImDispatch { - const admission = this.admitReceivedDirectMessage(message, fromHistory); + const admission = this.admitReceivedDirectMessage( + message, + fromHistory, + reportFailure, + ); const completed = admission.then(({ completion }) => completion); void completed.catch(() => undefined); return { @@ -1549,18 +1549,16 @@ export class DwsChannel extends PollingChannelBase { private async admitReceivedDirectMessage( message: DwsImMessage, fromHistory: boolean, + reportFailure: boolean, ): Promise<{ completion: Promise }> { if (!this.connected) return { completion: Promise.resolve() }; const key = messageKey(message); - if (this.isSelfMessage(message)) { - this.markProcessedMessage(key); - this.saveCursor(); + if (this.markSelfMessageProcessed(message)) { return { completion: Promise.resolve() }; } if ( - !fromHistory && - message.eventTime !== undefined && - message.eventTime < this.connectionStartedAt - 5_000 + this.isStaleLiveMessage(message, fromHistory) && + message.eventTime !== undefined ) { // A replayed direct message is left UNMARKED on purpose, for history // polling to pick up. That only works if polling will ever look @@ -1590,16 +1588,40 @@ export class DwsChannel extends PollingChannelBase { this.saveCursor(); return { completion: Promise.resolve() }; } - return this.admitDirectMessage({ kind: 'direct' }, message, key); + return this.admitDirectMessage( + { kind: 'direct' }, + message, + key, + reportFailure, + ); + } + + private markSelfMessageProcessed(message: DwsImMessage): boolean { + if (!this.isSelfMessage(message)) return false; + this.markProcessedMessage(messageKey(message)); + this.saveCursor(); + return true; + } + + private isStaleLiveMessage( + message: DwsImMessage, + fromHistory: boolean, + ): boolean { + return ( + !fromHistory && + message.eventTime !== undefined && + message.eventTime < this.connectionStartedAt - 5_000 + ); } private async admitDirectMessage( source: Extract, message: DwsImMessage, key: string, + reportFailure: boolean, ): Promise<{ completion: Promise }> { if (this.cursor.processedMessages.includes(key)) { - if (this.removePendingMessage(key)) this.saveCursor(); + this.removePersistedPendingMessage(key); return { completion: Promise.resolve() }; } if (!this.hasPendingMessage(key)) { @@ -1607,22 +1629,30 @@ export class DwsChannel extends PollingChannelBase { return { completion: Promise.resolve() }; } } - return { completion: this.scheduleDirectMessage(source, message, key) }; + return { + completion: this.scheduleDirectMessage( + source, + message, + key, + reportFailure, + ), + }; } private scheduleDirectMessage( source: Extract, message: DwsImMessage, key: string, + reportFailure: boolean, ): Promise { const queued = this.queuedDirectMessages.get(key); if (queued) return queued; if (this.cursor.processedMessages.includes(key)) { - if (this.removePendingMessage(key)) this.saveCursor(); + this.removePersistedPendingMessage(key); return Promise.resolve(); } const generation = this.lifecycleGeneration; - const task = Promise.resolve().then(async () => { + const dispatch = async () => { try { if (!this.connected || generation !== this.lifecycleGeneration) return; await this.dispatchImMessage(source, message, key); @@ -1631,14 +1661,35 @@ export class DwsChannel extends PollingChannelBase { this.queuedDirectMessages.delete(key); } } - }); + }; + const serializeConversation = this.config.dispatchMode === 'followup'; + const previous = serializeConversation + ? this.directConversationTails.get(message.conversationId) + : undefined; + const task = previous + ? previous.catch(() => undefined).then(dispatch) + : Promise.resolve().then(dispatch); this.queuedDirectMessages.set(key, task); - void task.catch((error: unknown) => { - this.logImError( - source, - error instanceof Error ? error : new Error(String(error)), - ); - }); + if (serializeConversation) { + this.directConversationTails.set(message.conversationId, task); + void task + .finally(() => { + if ( + this.directConversationTails.get(message.conversationId) === task + ) { + this.directConversationTails.delete(message.conversationId); + } + }) + .catch(() => undefined); + } + if (reportFailure) { + void task.catch((error: unknown) => { + this.logImError( + source, + error instanceof Error ? error : new Error(String(error)), + ); + }); + } return task; } @@ -1673,8 +1724,7 @@ export class DwsChannel extends PollingChannelBase { this.cursor.processedMessages.includes(key) && this.hasPendingMessage(key) ) { - this.removePendingMessage(key); - this.saveCursor(); + this.removePersistedPendingMessage(key); } } finally { if (this.processingMessages.get(key) === task) { @@ -1747,7 +1797,7 @@ export class DwsChannel extends PollingChannelBase { // Direct-message history may be unavailable, and ambient group // messages have no history fallback. `at` messages need no parking: // the pinned mention checkpoint re-fetches them. - await this.rememberPendingMessage(source, message); + await this.rememberFailedMessage(source, message); } // Under budget the throw propagates exactly as before, so redelivery // and concurrent-duplicate retry keep their contracts. Once the budget @@ -1866,6 +1916,37 @@ export class DwsChannel extends PollingChannelBase { return true; } + private async rememberFailedMessage( + source: PersistedPendingMessage['source'], + message: DwsImMessage, + ): Promise { + const key = messageKey(message); + if (this.hasPendingMessage(key)) return; + const generation = this.lifecycleGeneration; + while ( + this.connected && + generation === this.lifecycleGeneration && + (this.cursor.pendingMessages?.length ?? 0) >= MAX_PROCESSED_ITEMS + ) { + await new Promise((resolve) => { + this.pendingMessageCapacityWaiters.add(resolve); + }); + } + if (this.hasPendingMessage(key)) return; + const pending = this.cursor.pendingMessages ?? []; + pending.push({ source, message }); + this.cursor.pendingMessages = pending; + try { + this.saveCursor(); + } catch { + return; + } + } + + private removePersistedPendingMessage(key: string): void { + if (this.removePendingMessage(key)) this.saveCursor(); + } + private removePendingMessage(key: string): boolean { const pending = this.cursor.pendingMessages ?? []; const remaining = pending.filter( @@ -2022,7 +2103,11 @@ export class DwsChannel extends PollingChannelBase { if (signal.aborted || !this.connected) return; const key = messageKey(pending.message); if (pending.source.kind === 'direct') { - this.scheduleDirectMessage(pending.source, pending.message, key); + if (this.queuedDirectMessages.has(key)) continue; + if (this.queuedDirectMessages.size >= MAX_DIRECT_REPLAY_DISPATCHES) { + continue; + } + this.scheduleDirectMessage(pending.source, pending.message, key, true); continue; } try { diff --git a/packages/channels/dws/src/dws-client.test.ts b/packages/channels/dws/src/dws-client.test.ts index 6971306398a..47ef615ba7f 100644 --- a/packages/channels/dws/src/dws-client.test.ts +++ b/packages/channels/dws/src/dws-client.test.ts @@ -554,6 +554,49 @@ describe('DwsClient', () => { ); }); + it('does not report completion after direct admission fails', async () => { + let onLine!: (line: string) => void | Promise; + const eventStarter = vi.fn( + async (_executable, _args, lineHandler) => { + onLine = lineHandler; + return subscription(); + }, + ); + const client = new DwsClient( + { executable: '/opt/dws' }, + vi.fn(), + eventStarter, + ); + const onError = vi.fn(); + + await client.subscribeToIm( + { kind: 'direct' }, + vi.fn( + (): DwsImDispatch => ({ + admitted: Promise.reject(new Error('admission failed')), + completed: Promise.reject(new Error('completion failed')), + }), + ), + onError, + ); + + await expect( + onLine( + json({ + type: 'user_im_message_receive_o2o_all', + event_id: 'event-1', + message_id: 'message-1', + conversation_id: 'conversation-a', + content: 'request', + sender_open_dingtalk_id: 'user-a', + sender: 'User A', + }), + ), + ).rejects.toThrow('admission failed'); + await new Promise((resolve) => setTimeout(resolve, 0)); + expect(onError).not.toHaveBeenCalled(); + }); + it('observes completion failure while direct admission is pending', async () => { let onLine!: (line: string) => void | Promise; const eventStarter = vi.fn( From e9b74c4bd1a5935a2a91f0f18a380952a1cdcc4d Mon Sep 17 00:00:00 2001 From: qqqys <266654365+qqqys@users.noreply.github.com> Date: Wed, 2 Sep 2026 05:39:48 +0800 Subject: [PATCH 4/5] fix(dws): preserve direct message progress --- packages/channels/dws/src/dws-channel.test.ts | 116 ++++++++++++------ packages/channels/dws/src/dws-channel.ts | 89 +++++++++----- 2 files changed, 139 insertions(+), 66 deletions(-) diff --git a/packages/channels/dws/src/dws-channel.test.ts b/packages/channels/dws/src/dws-channel.test.ts index 15fde392013..0f8c9a75f02 100644 --- a/packages/channels/dws/src/dws-channel.test.ts +++ b/packages/channels/dws/src/dws-channel.test.ts @@ -1522,7 +1522,7 @@ describe('DwsChannel', () => { expect(bridge.prompt).toHaveBeenCalledTimes(2); }); - it('preserves followup order while the first message is classified', async () => { + it('preserves default steer order while the first message is classified', async () => { const client = new FakeDwsClient(); let finishClassification!: (result: { intent: 'none'; @@ -1542,13 +1542,16 @@ describe('DwsChannel', () => { }; const { bridge } = await readyPolicyChannel( client, - makeConfig({ dispatchMode: 'followup' }), + makeConfig(), 'classified-order-dws', { channelMemory: makeChannelMemory(), memoryIntentClassifier, }, ); + (bridge.cancelSession as ReturnType).mockResolvedValue( + undefined, + ); const firstDelivery = client.emit( 1, @@ -1635,25 +1638,20 @@ describe('DwsChannel', () => { await delivery; }); - it('releases capacity-blocked admission when disconnected', async () => { + it('rejects full-capacity admission without waiting', async () => { const client = new FakeDwsClient(); const channel = await readyChannel(client); channel.seedPendingMessages(5_000); - const delivery = client.emit( - 1, - message( - 'user_im_message_receive_o2o_all', - 'disconnect-capacity', - 'request', + await expect( + client.emit( + 1, + message('user_im_message_receive_o2o_all', 'full-capacity', 'request'), ), + ).rejects.toThrow( + 'DWS pending-message capacity is exhausted; retry later.', ); - await vi.waitFor(() => - expect(channel.pendingMessageCapacityWaiterCount()).toBe(1), - ); - channel.disconnect(); - await delivery; expect(channel.pendingMessageCapacityWaiterCount()).toBe(0); }); @@ -1726,33 +1724,29 @@ describe('DwsChannel', () => { } }); - it('applies backpressure instead of evicting an admitted message', async () => { + it('rejects admission instead of evicting a pending message', async () => { const client = new FakeDwsClient(); const channel = await readyChannel(client); channel.seedPendingMessages(5_000); - const delivery = client.emit( - 1, - message('user_im_message_receive_o2o_all', 'new-request', 'request', { - conversationId: 'conversation-new', - }), + await expect( + client.emit( + 1, + message('user_im_message_receive_o2o_all', 'new-request', 'request', { + conversationId: 'conversation-new', + }), + ), + ).rejects.toThrow( + 'DWS pending-message capacity is exhausted; retry later.', ); - await new Promise((resolve) => setTimeout(resolve, 0)); expect(channel.inbound).toEqual([]); expect(channel.pendingMessageIds()).toHaveLength(5_000); expect(channel.pendingMessageIds()[0]).toBe('parked-0'); expect(channel.pendingMessageIds()).not.toContain('new-request'); - - channel.releasePendingMessage('conversation-capacity', 'parked-0'); - await delivery; - expect(channel.inbound.map(({ messageId }) => messageId)).toEqual([ - 'new-request', - ]); - expect(channel.pendingMessageIds()).toHaveLength(4_999); }); - it('does not advance direct history past admission released by disconnect', async () => { + it('does not block or advance direct history at pending capacity', async () => { const client = new FakeDwsClient(); const channel = await readyChannel(client); const initialWatermark = channel.notificationWatermark()!; @@ -1779,14 +1773,10 @@ describe('DwsChannel', () => { .spyOn(Date, 'now') .mockReturnValue(initialWatermark + 10_000); try { - const poll = channel.poll(); - await vi.waitFor(() => - expect(channel.pendingMessageCapacityWaiterCount()).toBe(1), - ); - channel.disconnect(); - await poll; + await channel.poll(); expect(channel.notificationWatermark()).toBe(initialWatermark); + expect(channel.pendingMessageCapacityWaiterCount()).toBe(0); release(); await vi.waitFor(() => @@ -1814,6 +1804,54 @@ describe('DwsChannel', () => { } }); + it('drains failed replay capacity until direct history can resume', async () => { + const client = new FakeDwsClient(); + const channel = await readyChannel(client); + const initialWatermark = channel.notificationWatermark()!; + channel.seedPendingMessages(5_000, true); + channel.inboundError = new Error('agent unavailable'); + const victim = message( + 'user_im_message_receive_o2o_all', + 'history-capacity-victim', + 'request', + { + conversationId: 'history-victim', + eventTime: initialWatermark - 1_000, + }, + ); + client.directMessages = [victim]; + const now = vi + .spyOn(Date, 'now') + .mockReturnValue(initialWatermark + 10_000); + const stderr = vi.spyOn(process.stderr, 'write').mockReturnValue(true); + try { + for (const expectedAttempts of [16, 32, 48, 64]) { + await channel.poll(); + await vi.waitFor(() => + expect(channel.inboundAttempts).toBe(expectedAttempts), + ); + expect(channel.notificationWatermark()).toBe(initialWatermark); + } + + await channel.poll(); + await vi.waitFor(() => + expect(channel.pendingMessageIds().length).toBeLessThan(5_000), + ); + if (!channel.pendingMessageIds().includes('history-capacity-victim')) { + await channel.poll(); + } + await vi.waitFor(() => + expect(channel.pendingMessageIds()).toContain( + 'history-capacity-victim', + ), + ); + expect(channel.notificationWatermark()).toBeGreaterThan(initialWatermark); + } finally { + stderr.mockRestore(); + now.mockRestore(); + } + }); + it('does not acknowledge admission when cursor persistence fails', async () => { const client = new FakeDwsClient(); const channel = await readyChannel(client); @@ -2534,6 +2572,14 @@ describe('DwsChannel', () => { const logged = stderr.mock.calls.map((call) => String(call[0])).join(''); expect(logged).toContain('DWS message turn failed (attempt 1/5)'); expect(logged).not.toContain('failed to poll DWS direct-message history'); + await vi.waitFor(() => { + expect( + channel.queuedDirectMessage('cid-1\0failing-first'), + ).toBeUndefined(); + expect( + channel.queuedDirectMessage('cid-1\0waiting-second'), + ).toBeUndefined(); + }); channel.inboundError = undefined; await channel.poll(); diff --git a/packages/channels/dws/src/dws-channel.ts b/packages/channels/dws/src/dws-channel.ts index 977846f45a8..fed3d3e84ce 100644 --- a/packages/channels/dws/src/dws-channel.ts +++ b/packages/channels/dws/src/dws-channel.ts @@ -154,6 +154,11 @@ interface ImSubscriptionState { restartAttempts: number; } +interface DirectConversationTail { + started: Promise; + completed: Promise; +} + interface ActiveReaction { target: { conversationId: string; messageId: string }; sessionId: string; @@ -521,7 +526,11 @@ export class DwsChannel extends PollingChannelBase { private readonly notifiedSenderPairingNotifications = new Set(); private readonly processingMessages = new Map>(); private readonly queuedDirectMessages = new Map>(); - private readonly directConversationTails = new Map>(); + private readonly directConversationTails = new Map< + string, + DirectConversationTail + >(); + private readonly directMessageStartResolvers = new Map void>(); private readonly pendingMessageCapacityWaiters = new Set<() => void>(); private pollAbortController = new AbortController(); private lifecycleGeneration = 0; @@ -831,6 +840,8 @@ export class DwsChannel extends PollingChannelBase { } this.sessionReactionKeys.clear(); this.queuedDirectMessages.clear(); + for (const resolve of this.directMessageStartResolvers.values()) resolve(); + this.directMessageStartResolvers.clear(); this.directConversationTails.clear(); for (const resolve of this.pendingMessageCapacityWaiters) resolve(); this.pendingMessageCapacityWaiters.clear(); @@ -1662,26 +1673,35 @@ export class DwsChannel extends PollingChannelBase { } } }; - const serializeConversation = this.config.dispatchMode === 'followup'; - const previous = serializeConversation - ? this.directConversationTails.get(message.conversationId) - : undefined; - const task = previous - ? previous.catch(() => undefined).then(dispatch) + let resolveStarted!: () => void; + const started = new Promise((resolve) => { + resolveStarted = resolve; + }); + const previous = this.directConversationTails.get(message.conversationId); + // Live turns only need FIFO through ChannelBase registration; replay and + // followup turns must wait for the prior turn to finish. + const predecessor = + reportFailure || this.config.dispatchMode === 'followup' + ? previous?.completed + : previous?.started; + const task = predecessor + ? predecessor.catch(() => undefined).then(dispatch) : Promise.resolve().then(dispatch); + const tail = { started, completed: task }; this.queuedDirectMessages.set(key, task); - if (serializeConversation) { - this.directConversationTails.set(message.conversationId, task); - void task - .finally(() => { - if ( - this.directConversationTails.get(message.conversationId) === task - ) { - this.directConversationTails.delete(message.conversationId); - } - }) - .catch(() => undefined); - } + this.directMessageStartResolvers.set(key, resolveStarted); + this.directConversationTails.set(message.conversationId, tail); + void task + .finally(() => { + this.releaseDirectMessageStart( + message.conversationId, + message.messageId, + ); + if (this.directConversationTails.get(message.conversationId) === tail) { + this.directConversationTails.delete(message.conversationId); + } + }) + .catch(() => undefined); if (reportFailure) { void task.catch((error: unknown) => { this.logImError( @@ -1890,18 +1910,11 @@ export class DwsChannel extends PollingChannelBase { ): Promise { const key = messageKey(message); if (this.hasPendingMessage(key)) return true; - const generation = this.lifecycleGeneration; - while ( - this.connected && - generation === this.lifecycleGeneration && - (this.cursor.pendingMessages?.length ?? 0) >= MAX_PROCESSED_ITEMS - ) { - await new Promise((resolve) => { - this.pendingMessageCapacityWaiters.add(resolve); - }); - } - if (!this.connected || generation !== this.lifecycleGeneration) { - return false; + if (!this.connected) return false; + if ((this.cursor.pendingMessages?.length ?? 0) >= MAX_PROCESSED_ITEMS) { + throw new Error( + 'DWS pending-message capacity is exhausted; retry later.', + ); } if (this.hasPendingMessage(key)) return true; const pending = this.cursor.pendingMessages ?? []; @@ -2257,6 +2270,17 @@ export class DwsChannel extends PollingChannelBase { return `${conversationId}\0${messageId}`; } + private releaseDirectMessageStart( + conversationId: string, + messageId: string, + ): void { + const key = `${conversationId}\0${messageId}`; + const resolve = this.directMessageStartResolvers.get(key); + if (!resolve) return; + this.directMessageStartResolvers.delete(key); + resolve(); + } + private rememberInboundReactionTarget( chatId: string, messageId: string, @@ -2449,6 +2473,9 @@ export class DwsChannel extends PollingChannelBase { protected override onTaskLifecycle(event: ChannelTaskLifecycleEvent): void { if (event.type === 'started') { + if (event.messageId) { + this.releaseDirectMessageStart(event.chatId, event.messageId); + } this.startReaction(event.chatId, event.messageId, event.sessionId); return; } From 47852c6ef61fc3e52ec8e87be9004e48d4e1c7fa Mon Sep 17 00:00:00 2001 From: qqqys Date: Wed, 2 Sep 2026 17:49:32 +0800 Subject: [PATCH 5/5] fix(dws): scope the direct replay cap to replay dispatches The replay cap measured queuedDirectMessages.size, which in followup mode also counts the live per-conversation backlog: entries are added at schedule time and removed only when their own turn finishes, so N chained messages hold N entries while one turn runs. A single conversation's backlog of 16 could therefore keep every poll from replaying any parked failed direct message of any other conversation, and parked entries have no other redelivery surface. Count only replay-started dispatches against the cap, tracked in a map cleared on disconnect alongside queuedDirectMessages so replay-alone concurrency still stays at or below the cap across polls. Also add the missing test witnesses for the disconnect and tail cleanup introduced earlier in this branch: the conversation-tail reset on disconnect, the identity-guarded tail release in scheduleDirectMessage, and failed ambient parking when a capacity wait is released by disconnect. --- packages/channels/dws/src/dws-channel.test.ts | 212 ++++++++++++++++++ packages/channels/dws/src/dws-channel.ts | 21 +- 2 files changed, 231 insertions(+), 2 deletions(-) diff --git a/packages/channels/dws/src/dws-channel.test.ts b/packages/channels/dws/src/dws-channel.test.ts index 0f8c9a75f02..637c220ec0e 100644 --- a/packages/channels/dws/src/dws-channel.test.ts +++ b/packages/channels/dws/src/dws-channel.test.ts @@ -438,6 +438,24 @@ class TestableDwsChannel extends DwsChannel { ).queuedDirectMessages.get(key); } + directConversationTailIds(): string[] { + return [ + ...( + this as unknown as { + directConversationTails: Map; + } + ).directConversationTails.keys(), + ]; + } + + replayDirectDispatchCount(): number { + return ( + this as unknown as { + replayDirectDispatches: Map>; + } + ).replayDirectDispatches.size; + } + replaceQueuedDirectMessage(key: string, task: Promise): void { ( this as unknown as { @@ -470,6 +488,16 @@ class PolicyDwsChannel extends DwsChannel { ).queuedDirectMessages.size; } + directConversationTailIds(): string[] { + return [ + ...( + this as unknown as { + directConversationTails: Map; + } + ).directConversationTails.keys(), + ]; + } + documentSetSize(): number { return (this as unknown as { documentSet: Set }).documentSet.size; } @@ -1522,6 +1550,61 @@ describe('DwsChannel', () => { expect(bridge.prompt).toHaveBeenCalledTimes(2); }); + it('keeps direct-message order past the second turn and frees the tail', async () => { + const client = new FakeDwsClient(); + const { channel, bridge } = await readyPolicyChannel( + client, + makeConfig({ dispatchMode: 'followup' }), + ); + let releaseFirst!: () => void; + let releaseSecond!: () => void; + const firstBlocked = new Promise((resolve) => { + releaseFirst = resolve; + }); + const secondBlocked = new Promise((resolve) => { + releaseSecond = resolve; + }); + let promptCount = 0; + (bridge.prompt as ReturnType).mockImplementation(async () => { + promptCount += 1; + if (promptCount === 1) await firstBlocked; + if (promptCount === 2) await secondBlocked; + return 'response'; + }); + const emit = (messageId: string, content: string) => + client.emit( + 1, + message('user_im_message_receive_o2o_all', messageId, content, { + conversationId: 'conversation-ordered', + }), + ); + + const first = emit('ordered-1', 'first request'); + await vi.waitFor(() => expect(bridge.prompt).toHaveBeenCalledOnce()); + const second = emit('ordered-2', 'second request'); + + // Turn 1 settles while turn 2 becomes the conversation tail. Without the + // identity guard, turn 1's cleanup clears turn 2's entry, so the running + // conversation loses the tail that later turns must queue behind. + releaseFirst(); + await vi.waitFor(() => expect(bridge.prompt).toHaveBeenCalledTimes(2)); + expect(channel.directConversationTailIds()).toEqual([ + 'conversation-ordered', + ]); + + const third = emit('ordered-3', 'third request'); + await new Promise((resolve) => setTimeout(resolve, 0)); + expect(bridge.prompt).toHaveBeenCalledTimes(2); + + releaseSecond(); + await Promise.all([first, second, third]); + expect(bridge.prompt).toHaveBeenCalledTimes(3); + // A drained conversation must not keep a resolved tail around. + await vi.waitFor(() => + expect(channel.directConversationTailIds()).toEqual([]), + ); + }); + it('preserves default steer order while the first message is classified', async () => { const client = new FakeDwsClient(); let finishClassification!: (result: { @@ -1612,6 +1695,48 @@ describe('DwsChannel', () => { await vi.waitFor(() => expect(channel.pendingMessageIds()).toHaveLength(9)); }); + it('does not let a live followup backlog starve parked replay dispatches', async () => { + const client = new FakeDwsClient(); + const channel = await readyChannel( + client, + makeConfig({ dispatchMode: 'followup' }), + ); + // One parked failed direct message, in a conversation of its own. Replay is + // its only redelivery surface, so a starved replay pass strands it. + channel.seedPendingMessages(1); + let release!: () => void; + const blocked = new Promise((resolve) => { + release = resolve; + }); + channel.inboundHandler = async () => blocked; + + // A single conversation's backlog fills queuedDirectMessages to the cap + // while only the head turn actually runs. + const deliveries = Array.from({ length: 16 }, (_unused, index) => + client.emit( + 1, + message( + 'user_im_message_receive_o2o_all', + `backlog-${index}`, + `request ${index}`, + { conversationId: 'conversation-backlog' }, + ), + ), + ); + await vi.waitFor(() => expect(channel.inboundAttempts).toBe(1)); + expect(channel.pendingMessageIds()).toHaveLength(17); + + await channel.poll(); + + // The parked entry belongs to another conversation, so it dispatches + // immediately once the cap stops counting the backlog. + await vi.waitFor(() => expect(channel.inboundAttempts).toBe(2)); + expect(channel.replayDirectDispatchCount()).toBe(1); + + release(); + await Promise.all(deliveries); + }); + it('clears queued direct messages when disconnected', async () => { const client = new FakeDwsClient(); const channel = await readyChannel(client); @@ -1638,6 +1763,53 @@ describe('DwsChannel', () => { await delivery; }); + it('drops conversation tails on disconnect so reconnects do not chain onto them', async () => { + const client = new FakeDwsClient(); + const channel = await readyChannel( + client, + makeConfig({ dispatchMode: 'followup' }), + ); + let release!: () => void; + const blocked = new Promise((resolve) => { + release = resolve; + }); + channel.inboundHandler = async () => blocked; + + const stranded = client.emit( + 1, + message( + 'user_im_message_receive_o2o_all', + 'tail-before-disconnect', + 'first request', + { conversationId: 'conversation-tail' }, + ), + ); + await vi.waitFor(() => expect(channel.inboundAttempts).toBe(1)); + expect(channel.directConversationTailIds()).toEqual(['conversation-tail']); + + channel.disconnect(); + expect(channel.directConversationTailIds()).toEqual([]); + + await channel.connect(); + const afterReconnect = client.emit( + 1, + message( + 'user_im_message_receive_o2o_all', + 'tail-after-reconnect', + 'second request', + { conversationId: 'conversation-tail' }, + ), + ); + + // The pre-disconnect turn is still blocked. Had its tail survived, the new + // message would chain behind a promise from the previous lifecycle and + // never start. + await vi.waitFor(() => expect(channel.inboundAttempts).toBe(2)); + + release(); + await Promise.all([stranded, afterReconnect]); + }); + it('rejects full-capacity admission without waiting', async () => { const client = new FakeDwsClient(); const channel = await readyChannel(client); @@ -5436,6 +5608,46 @@ describe('DwsChannel', () => { ]); }); + it('retains a capacity-blocked failed ambient message across a disconnect', async () => { + const client = new FakeDwsClient(); + const channel = await readyChannel( + client, + makeConfig({ groups: { '*': { requireMention: false } } }), + ); + channel.seedPendingMessages(5_000); + channel.inboundError = new Error('agent unavailable'); + const event = message( + 'user_im_message_receive_group_all', + 'ambient-capacity-disconnect', + 'please retry this group request', + { conversationId: 'cid-group' }, + ); + + const delivery = client.emit(1, event); + const failedDelivery = + expect(delivery).rejects.toThrow('agent unavailable'); + await vi.waitFor(() => + expect(channel.pendingMessageCapacityWaiterCount()).toBe(1), + ); + + // disconnect() releases the capacity waiters. Ambient group messages have + // no history fallback, so parking must still happen on that exit. + channel.disconnect(); + await failedDelivery; + expect(channel.pendingMessageIds()).toContain( + 'ambient-capacity-disconnect', + ); + + channel.inboundError = undefined; + await channel.connect(); + await channel.poll(); + await vi.waitFor(() => + expect(channel.inbound).toContainEqual( + expect.objectContaining({ messageId: 'ambient-capacity-disconnect' }), + ), + ); + }); + it('replays a failed ambient group message after restart', async () => { const config = makeConfig({ groups: { '*': { requireMention: false } } }); const name = 'pending-ambient-group-dws'; diff --git a/packages/channels/dws/src/dws-channel.ts b/packages/channels/dws/src/dws-channel.ts index fed3d3e84ce..dc5626984a9 100644 --- a/packages/channels/dws/src/dws-channel.ts +++ b/packages/channels/dws/src/dws-channel.ts @@ -526,6 +526,9 @@ export class DwsChannel extends PollingChannelBase { private readonly notifiedSenderPairingNotifications = new Set(); private readonly processingMessages = new Map>(); private readonly queuedDirectMessages = new Map>(); + // Replay-started direct dispatches only. The cap must not be consumed by + // live or followup traffic, whose queue entries outlive their turn. + private readonly replayDirectDispatches = new Map>(); private readonly directConversationTails = new Map< string, DirectConversationTail @@ -840,6 +843,7 @@ export class DwsChannel extends PollingChannelBase { } this.sessionReactionKeys.clear(); this.queuedDirectMessages.clear(); + this.replayDirectDispatches.clear(); for (const resolve of this.directMessageStartResolvers.values()) resolve(); this.directMessageStartResolvers.clear(); this.directConversationTails.clear(); @@ -2117,10 +2121,23 @@ export class DwsChannel extends PollingChannelBase { const key = messageKey(pending.message); if (pending.source.kind === 'direct') { if (this.queuedDirectMessages.has(key)) continue; - if (this.queuedDirectMessages.size >= MAX_DIRECT_REPLAY_DISPATCHES) { + if (this.replayDirectDispatches.size >= MAX_DIRECT_REPLAY_DISPATCHES) { continue; } - this.scheduleDirectMessage(pending.source, pending.message, key, true); + const dispatched = this.scheduleDirectMessage( + pending.source, + pending.message, + key, + true, + ); + this.replayDirectDispatches.set(key, dispatched); + void dispatched + .finally(() => { + if (this.replayDirectDispatches.get(key) === dispatched) { + this.replayDirectDispatches.delete(key); + } + }) + .catch(() => undefined); continue; } try {