From 125f62784e2737ab196f46beac544a2f373339fa Mon Sep 17 00:00:00 2001 From: qqqys <266654365+qqqys@users.noreply.github.com> Date: Thu, 27 Aug 2026 18:02:38 +0800 Subject: [PATCH 1/5] fix(channels): recover DWS direct messages --- packages/channels/dws/src/dws-channel.test.ts | 123 ++++++++++++++++-- packages/channels/dws/src/dws-channel.ts | 31 +++-- 2 files changed, 133 insertions(+), 21 deletions(-) diff --git a/packages/channels/dws/src/dws-channel.test.ts b/packages/channels/dws/src/dws-channel.test.ts index 9628f9ba8b9..a53acb53b97 100644 --- a/packages/channels/dws/src/dws-channel.test.ts +++ b/packages/channels/dws/src/dws-channel.test.ts @@ -1101,6 +1101,32 @@ describe('DwsChannel', () => { ]); }); + it('lets polling recover a stale replayed direct message', async () => { + const client = new FakeDwsClient(); + const channel = await readyChannel(client); + const replay = message( + 'user_im_message_receive_o2o_all', + 'replayed-direct', + 'stale direct request', + { eventTime: Date.now() - 60_000 }, + ); + client.directMessages = [replay]; + + await client.emit(1, replay); + await channel.poll(); + + expect(channel.inbound).toEqual([ + expect.objectContaining({ + chatId: 'cid-1', + messageId: 'replayed-direct', + text: 'stale direct request', + }), + ]); + + await channel.poll(); + expect(channel.inbound).toHaveLength(1); + }); + // R4-4: the pullback above only rescues the replay if the poll that was in // flight when it happened does not finish by writing its own window's end // back over it. `checkpoint.endTime` is always past the replay's `eventTime`, @@ -1711,6 +1737,53 @@ describe('DwsChannel', () => { ]); }); + it('dispatches an ordinary direct message when the event stream misses it', async () => { + const client = new FakeDwsClient(); + const channel = await readyChannel(client); + client.directMessages = [ + message( + 'user_im_message_receive_o2o_all', + 'history-direct', + 'recover this request', + ), + ]; + + await channel.poll(); + + expect(channel.inbound).toEqual([ + expect.objectContaining({ + chatId: 'cid-1', + messageId: 'history-direct', + text: 'recover this request', + isGroup: false, + }), + ]); + }); + + it('applies sender pairing to a direct message recovered from history', async () => { + const client = new FakeDwsClient(); + const { channel, bridge } = await readyPolicyChannel( + client, + makeConfig({ senderPolicy: 'pairing' }), + ); + client.directMessages = [ + message( + 'user_im_message_receive_o2o_all', + 'history-pairing', + 'please help', + ), + ]; + + await channel.poll(); + + expect(bridge.prompt).not.toHaveBeenCalled(); + expect(client.sendImMessage).toHaveBeenCalledWith( + { kind: 'direct', openDingTalkId: 'open-alice' }, + expect.stringContaining('pairing code'), + expect.any(String), + ); + }); + it('dispatches a group mention when the event stream misses it', async () => { const client = new FakeDwsClient(); const channel = await readyChannel(client); @@ -2826,7 +2899,7 @@ describe('DwsChannel', () => { ); expect(bridge.prompt).toHaveBeenCalledOnce(); - expect(client.replyToImMessage).toHaveBeenCalledOnce(); + expect(client.sendImMessage).toHaveBeenCalledOnce(); }); it('removes a working reaction that finishes attaching after the task', async () => { @@ -3739,10 +3812,9 @@ describe('DwsChannel', () => { expect(channel.inboundAttempts).toBe(5); }); - // R12-1: a plain direct message whose turn throws had no redelivery path — - // the at-most-once event stream already consumed it, the DM history loop - // dispatches only document-mention notifications, and the pending queue - // parked only group sources. One transient failure lost it forever. + // R12-1: keep a local redelivery path when a direct-message turn throws. + // The event stream is at most once and remote history can be unavailable, + // so one transient failure must not lose the message forever. it('replays a failed direct message on the next poll', async () => { const client = new FakeDwsClient(); const channel = await readyChannel(client); @@ -4055,23 +4127,54 @@ describe('DwsChannel', () => { ]); }); - it('uses the originating message for an idempotent final reply', async () => { + it('sends an idempotent ordinary message for a final direct response', async () => { const client = new FakeDwsClient(); const channel = await readyChannel(client); + await client.emit( + 1, + message('user_im_message_receive_o2o_all', 'message-1', 'hello'), + ); channel.responseMessageId = 'message-1'; channel.responseSenderId = 'open-alice'; + await channel.respond('cid-1', 'final answer'); + const firstKey = client.sendImMessage.mock.calls[0]?.[2]; await channel.respond('cid-1', 'final answer'); - expect(client.replyToImMessage).toHaveBeenCalledWith( - 'cid-1', - 'message-1', - 'open-alice', + expect(client.sendImMessage).toHaveBeenNthCalledWith( + 1, + { kind: 'direct', openDingTalkId: 'open-alice' }, 'final answer', expect.stringMatching( /^[0-9a-f]{8}-[0-9a-f]{4}-5[0-9a-f]{3}-8[0-9a-f]{3}-[0-9a-f]{12}$/, ), ); + expect(client.sendImMessage.mock.calls[1]?.[2]).toBe(firstKey); + expect(client.replyToImMessage).not.toHaveBeenCalled(); + }); + + it('uses the originating message for a final group reply', async () => { + const client = new FakeDwsClient(); + const channel = await readyChannel(client); + await client.emit( + 0, + message('user_im_message_receive_at', 'message-1', '@Qwen hello', { + conversationId: 'group-1', + }), + ); + channel.responseMessageId = 'message-1'; + channel.responseSenderId = 'open-alice'; + + await channel.respond('group-1', 'final answer'); + + expect(client.replyToImMessage).toHaveBeenCalledWith( + 'group-1', + 'message-1', + 'open-alice', + 'final answer', + expect.any(String), + ); + expect(client.sendImMessage).not.toHaveBeenCalled(); }); // R1-7: the unknown-outcome swallow decides whether a finished task is diff --git a/packages/channels/dws/src/dws-channel.ts b/packages/channels/dws/src/dws-channel.ts index 92515edb348..1a4a3a9bfbc 100644 --- a/packages/channels/dws/src/dws-channel.ts +++ b/packages/channels/dws/src/dws-channel.ts @@ -1044,12 +1044,19 @@ export class DwsChannel extends PollingChannelBase { await this.sendMessage(chatId, text); return; } + const idempotencyKey = stableUuid( + `${this.name}\0${chatId}\0${messageId}\0${text}`, + ); + if (this.findImTarget(chatId)?.kind === 'direct') { + await this.sendImText(chatId, text, idempotencyKey); + return; + } await this.client.replyToImMessage( chatId, messageId, senderId, text, - stableUuid(`${this.name}\0${chatId}\0${messageId}\0${text}`), + idempotencyKey, ); } @@ -1130,11 +1137,14 @@ export class DwsChannel extends PollingChannelBase { continue; } const notification = parseDocumentMentionNotification(message.content); - if (!notification) continue; - await this.processDocumentNotification(message, key, notification); + if (notification) { + await this.processDocumentNotification(message, key, notification); + } else { + await this.handleImMessage({ kind: 'direct' }, message, true); + } } if (this.notificationWatermarkPulledBack) { - // R4-4: a stale document notification replayed while this window's + // R4-4: a stale direct message replayed while this window's // fetch was in flight, and `handleImMessage` 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: @@ -1458,13 +1468,13 @@ export class DwsChannel extends PollingChannelBase { message.eventTime !== undefined && message.eventTime < this.connectionStartedAt - 5_000 ) { - if (!parseDocumentMentionNotification(message.content)) { + if (source.kind !== 'direct') { this.markProcessedMessage(messageKey(message)); this.saveCursor(); return; } - // A replayed document notification is left UNMARKED on purpose, for - // history polling to pick up. That only works if polling will ever look + // 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 // `connectionStartedAt` and the query window opens at `watermark − 5s` // — exactly this branch's drop boundary — so every message this branch @@ -1582,10 +1592,9 @@ export class DwsChannel extends PollingChannelBase { } catch (error) { if (source.kind !== 'at') { // R12-1: park failed direct messages next to ambient group ones. - // The DM history loop dispatches document-mention notifications - // only, so nothing else ever re-drove a plain DM whose turn threw - // once. `at` messages need no parking: the pinned mention - // checkpoint re-fetches them. + // 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); } // Under budget the throw propagates exactly as before, so redelivery From 8295930d281179da9d3b18e6cced7ecbdc551ca6 Mon Sep 17 00:00:00 2001 From: qqqys <266654365+qqqys@users.noreply.github.com> Date: Thu, 27 Aug 2026 13:54:33 +0000 Subject: [PATCH 2/5] fix(channels): harden DWS direct-message recovery per review (#10274) --- docs/users/features/channels/dws.md | 2 + packages/channels/dws/src/dws-channel.test.ts | 106 ++++++++++++++---- packages/channels/dws/src/dws-channel.ts | 24 ++-- 3 files changed, 99 insertions(+), 33 deletions(-) diff --git a/docs/users/features/channels/dws.md b/docs/users/features/channels/dws.md index 26f775a9c6b..50a1ada6d94 100644 --- a/docs/users/features/channels/dws.md +++ b/docs/users/features/channels/dws.md @@ -77,6 +77,8 @@ qwen channel pairing approve dws-work CODE Group mentions use the real-time personal event stream first. The channel also checks recent `@` message history every five seconds, so mentions from external groups are recovered when DingTalk omits them from the personal event stream. Messages are deduplicated by conversation and message ID across both paths. +Ordinary direct messages are recovered the same way: a five-second history check re-drives any direct message the real-time stream omitted, deduplicated by conversation and message ID across both paths. + When a message quotes another DingTalk message, the quoted text is included as reply context for the agent on both the real-time and history fallback paths. ## Document Mentions diff --git a/packages/channels/dws/src/dws-channel.test.ts b/packages/channels/dws/src/dws-channel.test.ts index a53acb53b97..9364dbf59db 100644 --- a/packages/channels/dws/src/dws-channel.test.ts +++ b/packages/channels/dws/src/dws-channel.test.ts @@ -7,6 +7,7 @@ import { mkdtempSync, rmSync } from 'node:fs'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; +import process from 'node:process'; import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { PairingStore, @@ -1102,29 +1103,37 @@ describe('DwsChannel', () => { }); it('lets polling recover a stale replayed direct message', async () => { - const client = new FakeDwsClient(); - const channel = await readyChannel(client); - const replay = message( - 'user_im_message_receive_o2o_all', - 'replayed-direct', - 'stale direct request', - { eventTime: Date.now() - 60_000 }, - ); - client.directMessages = [replay]; + const stderr = vi.spyOn(process.stderr, 'write').mockReturnValue(true); + try { + const client = new FakeDwsClient(); + const channel = await readyChannel(client); + const replay = message( + 'user_im_message_receive_o2o_all', + 'replayed-direct', + 'stale direct request', + { eventTime: Date.now() - 60_000 }, + ); + client.directMessages = [replay]; - await client.emit(1, replay); - await channel.poll(); + await client.emit(1, replay); + expect(stderr).toHaveBeenCalledWith( + expect.stringContaining('parked a stale direct message'), + ); + await channel.poll(); - expect(channel.inbound).toEqual([ - expect.objectContaining({ - chatId: 'cid-1', - messageId: 'replayed-direct', - text: 'stale direct request', - }), - ]); + expect(channel.inbound).toEqual([ + expect.objectContaining({ + chatId: 'cid-1', + messageId: 'replayed-direct', + text: 'stale direct request', + }), + ]); - await channel.poll(); - expect(channel.inbound).toHaveLength(1); + await channel.poll(); + expect(channel.inbound).toHaveLength(1); + } finally { + stderr.mockRestore(); + } }); // R4-4: the pullback above only rescues the replay if the poll that was in @@ -1760,6 +1769,26 @@ describe('DwsChannel', () => { ]); }); + // 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 + // a duplicate agent turn. + it('deduplicates a direct message delivered by the live stream and history', async () => { + const client = new FakeDwsClient(); + const channel = await readyChannel(client); + const event = message( + 'user_im_message_receive_o2o_all', + 'live-and-history', + 'hello once', + ); + + await client.emit(1, event); + client.directMessages = [event]; + await channel.poll(); + + expect(channel.inbound).toHaveLength(1); + }); + it('applies sender pairing to a direct message recovered from history', async () => { const client = new FakeDwsClient(); const { channel, bridge } = await readyPolicyChannel( @@ -3886,6 +3915,39 @@ describe('DwsChannel', () => { expect(channel.inboundAttempts).toBe(5); }); + // R1-1: `replayPendingMessages` already re-drives a parked direct message + // every poll, so the history dispatch must skip it. Driving it from both + // surfaces spent the shared retry budget twice per poll, dropping a message + // after a transient outage barely longer than two poll intervals. + it('spends one retry per poll on a failed direct message also in history', async () => { + const client = new FakeDwsClient(); + const channel = await readyChannel(client); + channel.inboundError = new Error('agent unavailable'); + const event = message( + 'user_im_message_receive_o2o_all', + 'direct-history-retry', + 'please retry this direct request', + ); + client.directMessages = [event]; + + await expect(client.emit(1, event)).rejects.toThrow('agent unavailable'); + await channel.poll(); + await channel.poll(); + + expect(channel.inboundAttempts).toBe(3); + + channel.inboundError = undefined; + await channel.poll(); + + expect(channel.inboundAttempts).toBe(4); + expect(channel.inbound).toEqual([ + expect.objectContaining({ + chatId: 'cid-1', + messageId: 'direct-history-retry', + }), + ]); + }); + // R4-1: the budget above was wired into the mention path only. A document // notification whose turn throws escaped `pollOnce`'s sorted loop, so // nothing was marked processed, the checkpoint and watermark never advanced, @@ -4172,7 +4234,9 @@ describe('DwsChannel', () => { 'message-1', 'open-alice', 'final answer', - expect.any(String), + expect.stringMatching( + /^[0-9a-f]{8}-[0-9a-f]{4}-5[0-9a-f]{3}-8[0-9a-f]{3}-[0-9a-f]{12}$/, + ), ); expect(client.sendImMessage).not.toHaveBeenCalled(); }); diff --git a/packages/channels/dws/src/dws-channel.ts b/packages/channels/dws/src/dws-channel.ts index 1a4a3a9bfbc..d949fb7b230 100644 --- a/packages/channels/dws/src/dws-channel.ts +++ b/packages/channels/dws/src/dws-channel.ts @@ -1128,20 +1128,17 @@ export class DwsChannel extends PollingChannelBase { ); for (const message of page.messages) { if (signal.aborted || !this.connected) return; - if (this.isSelfMessage(message)) { - this.markProcessedMessage(messageKey(message)); - continue; - } - const key = messageKey(message); - if (this.cursor.processedMessages.includes(key)) { + // A parked message is already re-driven every poll by + // `replayPendingMessages`; dispatching it here too would spend the + // shared retry budget twice per poll. + if ( + (this.cursor.pendingMessages ?? []).some( + (pending) => messageKey(pending.message) === messageKey(message), + ) + ) { continue; } - const notification = parseDocumentMentionNotification(message.content); - if (notification) { - await this.processDocumentNotification(message, key, notification); - } else { - await this.handleImMessage({ kind: 'direct' }, message, true); - } + await this.handleImMessage({ kind: 'direct' }, message, true); } if (this.notificationWatermarkPulledBack) { // R4-4: a stale direct message replayed while this window's @@ -1495,6 +1492,9 @@ export class DwsChannel extends PollingChannelBase { // a replay no window will ever cover. Drop the checkpoint here too, so // the pullback survives regardless of when it arrived. this.cursor.notificationCheckpoint = undefined; + process.stderr.write( + `[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; } From 75ae02595715982bb968ac4191be1b9954882ed5 Mon Sep 17 00:00:00 2001 From: qqqys <266654365+qqqys@users.noreply.github.com> Date: Thu, 27 Aug 2026 17:30:45 +0000 Subject: [PATCH 3/5] fix(channels): contain DWS history-dispatch costs and failures per review (#10274) --- packages/channels/dws/src/dws-channel.test.ts | 120 ++++++++++++++++++ packages/channels/dws/src/dws-channel.ts | 25 +++- 2 files changed, 143 insertions(+), 2 deletions(-) diff --git a/packages/channels/dws/src/dws-channel.test.ts b/packages/channels/dws/src/dws-channel.test.ts index 9364dbf59db..edb193e0664 100644 --- a/packages/channels/dws/src/dws-channel.test.ts +++ b/packages/channels/dws/src/dws-channel.test.ts @@ -1789,6 +1789,126 @@ describe('DwsChannel', () => { expect(channel.inbound).toHaveLength(1); }); + // 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. + it('does not dispatch self-sent messages recovered from direct-message history', async () => { + const client = new FakeDwsClient(); + const channel = await readyChannel(client); + client.directMessages = [ + message('user_im_message_receive_o2o_all', 'own-reply', 'bot text', { + senderId: 'open-self', + }), + ]; + + await channel.poll(); + expect(channel.inbound).toEqual([]); + + await channel.poll(); + expect(channel.inbound).toEqual([]); + }); + + // 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 + // 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(); + const channel = await readyChannel(client); + client.directMessages = [ + message('user_im_message_receive_o2o_all', 'own-reply-cost', 'bot text', { + senderId: 'open-self', + }), + ]; + await channel.poll(); + const saveCursor = vi.spyOn( + channel as unknown as { saveCursor: () => void }, + 'saveCursor', + ); + + await channel.poll(); + + expect(channel.inbound).toEqual([]); + expect(saveCursor).toHaveBeenCalledTimes(1); + }); + + // R2-2: an inbound-turn failure during history dispatch escaped into the + // fetch catch, logged an agent-side failure as "failed to poll DWS + // direct-message history", and aborted the page. A failed direct message + // is parked for replay, so the page can keep moving instead. + it('keeps the page moving when a direct-message turn fails mid-window', 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 now = Date.now(); + client.directMessages = [ + message('user_im_message_receive_o2o_all', 'failing-first', 'first', { + eventTime: now - 1, + }), + message('user_im_message_receive_o2o_all', 'waiting-second', 'second', { + eventTime: now, + }), + ]; + + await channel.poll(); + + 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', + ]); + } finally { + stderr.mockRestore(); + } + }); + + // 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 () => { + vi.useFakeTimers(); + try { + const client = new FakeDwsClient(); + const channel = await readyChannel(client); + client.directMessages = [ + message( + 'user_im_message_receive_o2o_all', + 'stuck-notification', + documentMentionCard('doc-stuck', 'd'.repeat(45)), + { eventTime: Date.now() }, + ), + ]; + channel.inboundHandler = async () => { + throw new Error('agent unavailable'); + }; + + for (let round = 0; round < 5; round += 1) { + await channel.poll(); + await vi.advanceTimersByTimeAsync(6_000); + } + + expect(channel.inboundAttempts).toBe(5); + } finally { + vi.useRealTimers(); + } + }); + it('applies sender pairing to a direct message recovered from history', async () => { const client = new FakeDwsClient(); const { channel, bridge } = await readyPolicyChannel( diff --git a/packages/channels/dws/src/dws-channel.ts b/packages/channels/dws/src/dws-channel.ts index d949fb7b230..7717b59fd50 100644 --- a/packages/channels/dws/src/dws-channel.ts +++ b/packages/channels/dws/src/dws-channel.ts @@ -1128,17 +1128,38 @@ export class DwsChannel extends PollingChannelBase { ); for (const message of page.messages) { if (signal.aborted || !this.connected) return; + const key = messageKey(message); + if (this.cursor.processedMessages.includes(key)) { + continue; + } // A parked message is already re-driven every poll by // `replayPendingMessages`; dispatching it here too would spend the // shared retry budget twice per poll. if ( (this.cursor.pendingMessages ?? []).some( - (pending) => messageKey(pending.message) === messageKey(message), + (pending) => messageKey(pending.message) === key, ) ) { continue; } - await this.handleImMessage({ kind: 'direct' }, message, true); + try { + await this.handleImMessage({ kind: 'direct' }, message, true); + } 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.cursor.pendingMessages ?? []).some( + (pending) => messageKey(pending.message) === 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`, + ); + } } if (this.notificationWatermarkPulledBack) { // R4-4: a stale direct message replayed while this window's From d14b9e8b23602a035d8a1035c292d5539c53ad18 Mon Sep 17 00:00:00 2001 From: qwen-code-ci-bot Date: Thu, 27 Aug 2026 21:05:40 +0000 Subject: [PATCH 4/5] fix(channels): close the DWS in-flight double-spend per review (#10274) --- packages/channels/dws/src/dws-channel.test.ts | 128 ++++++++++++++++++ packages/channels/dws/src/dws-channel.ts | 38 +++--- 2 files changed, 150 insertions(+), 16 deletions(-) diff --git a/packages/channels/dws/src/dws-channel.test.ts b/packages/channels/dws/src/dws-channel.test.ts index edb193e0664..a60b8a06d36 100644 --- a/packages/channels/dws/src/dws-channel.test.ts +++ b/packages/channels/dws/src/dws-channel.test.ts @@ -1136,6 +1136,27 @@ describe('DwsChannel', () => { } }); + // R3-2: the stale-replay rescue is deliberately direct-only; every other + // source is still marked processed and dropped here. A mention replayed + // long after it was sent must not be parked and re-driven as a fresh turn. + it('still drops a stale replayed non-direct message', async () => { + const client = new FakeDwsClient(); + const channel = await readyChannel(client); + const replay = message( + 'user_im_message_receive_at', + 'stale-at-replay', + '@Qwen stale mention', + { eventTime: Date.now() - 60_000 }, + ); + client.directMessages = [replay]; + + await client.emit(0, replay); + expect(channel.inbound).toEqual([]); + + await channel.poll(); + expect(channel.inbound).toEqual([]); + }); + // R4-4: the pullback above only rescues the replay if the poll that was in // flight when it happened does not finish by writing its own window's end // back over it. `checkpoint.endTime` is always past the replay's `eventTime`, @@ -4068,6 +4089,113 @@ 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 + // 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. + it('spends one retry per poll when the live turn fails while history waits on it', async () => { + const client = new FakeDwsClient(); + const channel = await readyChannel(client); + const event = message( + 'user_im_message_receive_o2o_all', + 'direct-inflight-retry', + 'please retry this direct request', + ); + client.directMessages = [event]; + + let turnEntered!: () => void; + let failTurn!: () => void; + const entered = new Promise((resolve) => { + turnEntered = resolve; + }); + const failing = new Promise((resolve) => { + failTurn = resolve; + }); + channel.inboundHandler = async () => { + turnEntered(); + await failing; + throw new Error('agent unavailable'); + }; + + const liveTurn = client.emit(1, event).then( + () => undefined, + () => undefined, + ); + await entered; + const poll = channel.poll(); + // Let the poll's history dispatch reach the in-flight wait before the + // live turn is allowed to fail. + await new Promise((resolve) => setTimeout(resolve, 0)); + failTurn(); + await Promise.all([liveTurn, poll]); + + expect(channel.inboundAttempts).toBe(1); + + channel.inboundHandler = undefined; + await channel.poll(); + + expect(channel.inboundAttempts).toBe(2); + expect(channel.inbound).toEqual([ + expect.objectContaining({ + chatId: 'cid-1', + messageId: 'direct-inflight-retry', + }), + ]); + }); + + // The gated in-flight re-check rethrows instead of returning: + // `replayPendingMessages` deletes the parked entry after any normal + // return, so a redelivered duplicate whose turn fails while the replay + // waits on it must not make the replay drop the parking. + it('keeps a parked message parked when its replay waits on a failed duplicate turn', async () => { + const client = new FakeDwsClient(); + const channel = await readyChannel(client); + const event = message( + 'user_im_message_receive_o2o_all', + 'direct-parked-duplicate', + 'please retry this direct request', + ); + channel.inboundError = new Error('agent unavailable'); + + await expect(client.emit(1, event)).rejects.toThrow('agent unavailable'); + + let releaseDuplicate!: () => void; + const held = new Promise((resolve) => { + releaseDuplicate = resolve; + }); + channel.inboundHandler = async () => { + await held; + throw new Error('agent unavailable'); + }; + channel.inboundError = undefined; + const duplicateTurn = client.emit(1, event).then( + () => undefined, + () => undefined, + ); + await new Promise((resolve) => setTimeout(resolve, 0)); + const poll = channel.poll(); + await new Promise((resolve) => setTimeout(resolve, 0)); + releaseDuplicate(); + await Promise.all([duplicateTurn, poll]); + + // The duplicate turn spent attempt 2; the replay spent nothing and must + // have kept the parked entry. + expect(channel.inboundAttempts).toBe(2); + + channel.inboundHandler = undefined; + await channel.poll(); + + expect(channel.inboundAttempts).toBe(3); + expect(channel.inbound).toEqual([ + expect.objectContaining({ + chatId: 'cid-1', + messageId: 'direct-parked-duplicate', + }), + ]); + }); + // R4-1: the budget above was wired into the mention path only. A document // notification whose turn throws escaped `pollOnce`'s sorted loop, so // nothing was marked processed, the checkpoint and watermark never advanced, diff --git a/packages/channels/dws/src/dws-channel.ts b/packages/channels/dws/src/dws-channel.ts index 7717b59fd50..885c8ddc14c 100644 --- a/packages/channels/dws/src/dws-channel.ts +++ b/packages/channels/dws/src/dws-channel.ts @@ -1135,13 +1135,7 @@ export class DwsChannel extends PollingChannelBase { // A parked message is already re-driven every poll by // `replayPendingMessages`; dispatching it here too would spend the // shared retry budget twice per poll. - if ( - (this.cursor.pendingMessages ?? []).some( - (pending) => messageKey(pending.message) === key, - ) - ) { - continue; - } + if (this.hasPendingMessage(key)) continue; try { await this.handleImMessage({ kind: 'direct' }, message, true); } catch (error) { @@ -1149,13 +1143,7 @@ export class DwsChannel extends PollingChannelBase { // 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.cursor.pendingMessages ?? []).some( - (pending) => messageKey(pending.message) === key, - ) - ) { - throw error; - } + 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`, ); @@ -1535,12 +1523,24 @@ export class DwsChannel extends PollingChannelBase { return; } const key = messageKey(message); + let waitedOnInFlight = false; + let inFlightError: unknown; while (true) { const existing = this.processingMessages.get(key); if (!existing) break; - await existing.catch(() => undefined); + waitedOnInFlight = true; + inFlightError = await existing.then( + () => undefined, + (error: unknown) => error, + ); } if (this.cursor.processedMessages.includes(key)) return; + // The in-flight turn failed and parked the message while this caller + // waited; replay already re-drives parked entries every poll, so a new + // turn here would spend the retry budget twice in one poll. Rethrow + // instead of returning so `replayPendingMessages` keeps the entry — a + // normal return there deletes it. + if (waitedOnInFlight && this.hasPendingMessage(key)) throw inFlightError; const task = this.processImMessage(source, message, key); this.processingMessages.set(key, task); try { @@ -1708,8 +1708,8 @@ export class DwsChannel extends PollingChannelBase { message: DwsImMessage, ): void { const key = messageKey(message); + if (this.hasPendingMessage(key)) return; const pending = this.cursor.pendingMessages ?? []; - if (pending.some((item) => messageKey(item.message) === key)) return; while (pending.length >= MAX_PROCESSED_ITEMS) { const dropped = pending.shift(); if (!dropped) break; @@ -1920,6 +1920,12 @@ export class DwsChannel extends PollingChannelBase { } } + private hasPendingMessage(key: string): boolean { + return (this.cursor.pendingMessages ?? []).some( + (pending) => messageKey(pending.message) === key, + ); + } + private hasPendingDocumentNotification(notificationKey: string): boolean { return (this.cursor.pendingDocumentNotifications ?? []).some( (pending) => documentNotificationKey(pending) === notificationKey, From f403117d4cefbd132fe2e66d8a05b708eb4ecf42 Mon Sep 17 00:00:00 2001 From: qwen-code-ci-bot Date: Fri, 28 Aug 2026 02:31:42 +0000 Subject: [PATCH 5/5] test(channels): pin the DWS stale non-direct drop mark per review (#10274) --- packages/channels/dws/src/dws-channel.test.ts | 64 ++++++++++++++----- 1 file changed, 49 insertions(+), 15 deletions(-) diff --git a/packages/channels/dws/src/dws-channel.test.ts b/packages/channels/dws/src/dws-channel.test.ts index a60b8a06d36..dc22b8058f6 100644 --- a/packages/channels/dws/src/dws-channel.test.ts +++ b/packages/channels/dws/src/dws-channel.test.ts @@ -1137,24 +1137,58 @@ describe('DwsChannel', () => { }); // R3-2: the stale-replay rescue is deliberately direct-only; every other - // source is still marked processed and dropped here. A mention replayed - // long after it was sent must not be parked and re-driven as a fresh turn. + // source is still marked processed and dropped here. The mark is what keeps + // a replayed mention from becoming a fresh turn after a restart: the + // persisted mention watermark re-opens the overlap window over the downtime + // gap — which can lie entirely before the new connection's drop boundary — + // so history polling re-fetches the replay and only the mark drops it. it('still drops a stale replayed non-direct message', async () => { - const client = new FakeDwsClient(); - const channel = await readyChannel(client); - const replay = message( - 'user_im_message_receive_at', - 'stale-at-replay', - '@Qwen stale mention', - { eventTime: Date.now() - 60_000 }, - ); - client.directMessages = [replay]; + vi.useFakeTimers(); + try { + const name = 'stale-at-replay-dws'; + const firstClient = new FakeDwsClient(); + const first = await readyChannel(firstClient, makeConfig(), name); + firstClient.mentionedMessages = [ + message('user_im_message_receive_at', 'seed-mention', '@Qwen seed'), + ]; + await first.poll(); + expect(first.inbound).toHaveLength(1); + first.disconnect(); - await client.emit(0, replay); - expect(channel.inbound).toEqual([]); + await vi.advanceTimersByTimeAsync(30_000); - await channel.poll(); - expect(channel.inbound).toEqual([]); + const secondClient = new FakeDwsClient(); + const second = await readyChannel(secondClient, makeConfig(), name); + const replay = message( + 'user_im_message_receive_at', + 'stale-at-replay', + '@Qwen stale mention', + { eventTime: Date.now() - 15_000 }, + ); + secondClient.mentionedMessages = [replay]; + const windows: Array<[number, number]> = []; + const listMentionedMessages = + secondClient.listMentionedMessages.getMockImplementation(); + secondClient.listMentionedMessages.mockImplementation( + async (startTime, endTime, signal, cursor) => { + windows.push([startTime, endTime]); + return listMentionedMessages!(startTime, endTime, signal, cursor); + }, + ); + + await secondClient.emit(0, replay); + expect(second.inbound).toEqual([]); + + await second.poll(); + expect(second.inbound).toEqual([]); + // The restored watermark's window has to actually reach back over the + // replay — asserting only on `inbound` would pass on a watermark that + // restarted at the second connect and silently vacate the witness. + expect(windows[0][0]).toBeLessThanOrEqual(replay.eventTime!); + expect(windows[0][1]).toBeGreaterThanOrEqual(replay.eventTime!); + } finally { + vi.useRealTimers(); + } }); // R4-4: the pullback above only rescues the replay if the poll that was in