fix(channels): recover DWS direct messages

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

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,
@@ -1101,6 +1102,40 @@ describe('DwsChannel', () => {
     ]);
   });
 
+  it('lets polling recover a stale replayed direct message', async () => {
+    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);
+      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',
+        }),
+      ]);
+
+      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
   // 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 +1746,73 @@ 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,
+      }),
+    ]);
+  });
+
+  // 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(
+      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 +2928,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 +3841,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);
@@ -3814,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,
@@ -4055,23 +4189,56 @@ 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.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(
-      'cid-1',
+      'group-1',
       'message-1',
       '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).not.toHaveBeenCalled();
   });
 
   // R1-7: the unknown-outcome swallow decides whether a finished task is

packages/channels/dws/src/dws-channel.ts

@@ -1044,12 +1044,19 @@ export class DwsChannel extends PollingChannelBase<DwsCursor> {
       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,
     );
   }
 
@@ -1121,20 +1128,20 @@ export class DwsChannel extends PollingChannelBase<DwsCursor> {
       );
       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) continue;
-        await this.processDocumentNotification(message, key, notification);
+        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 +1465,13 @@ export class DwsChannel extends PollingChannelBase<DwsCursor> {
       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
@@ -1485,6 +1492,9 @@ export class DwsChannel extends PollingChannelBase<DwsCursor> {
       // 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;
     }
@@ -1582,10 +1592,9 @@ export class DwsChannel extends PollingChannelBase<DwsCursor> {
     } 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