diff --git a/packages/channels/dws/src/dws-channel.test.ts b/packages/channels/dws/src/dws-channel.test.ts
index b0c1f28e907..637c220ec0e 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';
@@ -22,6 +23,7 @@ import {
   type DwsClientLike,
   type DwsCommandRunner,
   type DwsIdentity,
+  type DwsImMessageResult,
   type DwsImMessage,
   type DwsImSource,
   type DwsImTarget,
@@ -62,6 +64,23 @@ function makeBridge(): ChannelAgentBridge {
   } as unknown as ChannelAgentBridge;
 }
 
+function makeChannelMemory(): NonNullable<ChannelBaseOptions['channelMemory']> {
+  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,
@@ -152,7 +171,7 @@ class FakeSubscription implements DwsEventSubscription {
 
 interface FakeStream {
   source: DwsImSource;
-  onMessage: (message: DwsImMessage) => void | Promise<void>;
+  onMessage: (message: DwsImMessage) => DwsImMessageResult;
   onError: (error: Error) => void;
   subscription: FakeSubscription;
 }
@@ -227,7 +246,7 @@ class FakeDwsClient implements DwsClientLike {
 
   async subscribeToIm(
     source: DwsImSource,
-    onMessage: (message: DwsImMessage) => void | Promise<void>,
+    onMessage: (message: DwsImMessage) => DwsImMessageResult,
     onError: (error: Error) => void,
   ): Promise<DwsEventSubscription> {
     const subscription = new FakeSubscription();
@@ -238,7 +257,11 @@ class FakeDwsClient implements DwsClientLike {
   async emit(sourceIndex: number, event: DwsImMessage): Promise<void> {
     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.admitted;
+      await result.completed;
+    } else await result;
   }
 }
 
@@ -246,6 +269,7 @@ class TestableDwsChannel extends DwsChannel {
   inbound: Envelope[] = [];
   inboundError?: Error;
   inboundHandler?: (envelope: Envelope) => Promise<void>;
+  nextCursorSaveError?: Error;
   responseMessageId?: string;
   responseSenderId?: string;
   responseThreadId?: string;
@@ -256,6 +280,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<void> {
@@ -321,6 +354,45 @@ class TestableDwsChannel extends DwsChannel {
     return this.cursor.mentionWatermark;
   }
 
+  pendingMessageIds(): string[] {
+    return (this.cursor.pendingMessages ?? []).map(
+      ({ message }) => message.messageId,
+    );
+  }
+
+  seedPendingMessages(count: number, separateConversations = false): 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: separateConversations
+              ? `conversation-capacity-${index}`
+              : 'conversation-capacity',
+          },
+        ),
+      }),
+    );
+    this.saveCursor();
+  }
+
+  releasePendingMessage(conversationId: string, messageId: string): void {
+    const removePendingMessage = (
+      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<string> {
     return this.router.resolve(this.name, 'alice', 'doc-1', 'comment-1');
   }
@@ -349,6 +421,48 @@ 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<void> | undefined {
+    return (
+      this as unknown as {
+        queuedDirectMessages: Map<string, Promise<void>>;
+      }
+    ).queuedDirectMessages.get(key);
+  }
+
+  directConversationTailIds(): string[] {
+    return [
+      ...(
+        this as unknown as {
+          directConversationTails: Map<string, unknown>;
+        }
+      ).directConversationTails.keys(),
+    ];
+  }
+
+  replayDirectDispatchCount(): number {
+    return (
+      this as unknown as {
+        replayDirectDispatches: Map<string, Promise<void>>;
+      }
+    ).replayDirectDispatches.size;
+  }
+
+  replaceQueuedDirectMessage(key: string, task: Promise<void>): void {
+    (
+      this as unknown as {
+        queuedDirectMessages: Map<string, Promise<void>>;
+      }
+    ).queuedDirectMessages.set(key, task);
+  }
 }
 
 class PolicyDwsChannel extends DwsChannel {
@@ -366,6 +480,24 @@ class PolicyDwsChannel extends DwsChannel {
     return this.cursor.pendingDocumentNotifications ?? [];
   }
 
+  queuedDirectMessageCount(): number {
+    return (
+      this as unknown as {
+        queuedDirectMessages: Map<string, Promise<void>>;
+      }
+    ).queuedDirectMessages.size;
+  }
+
+  directConversationTailIds(): string[] {
+    return [
+      ...(
+        this as unknown as {
+          directConversationTails: Map<string, unknown>;
+        }
+      ).directConversationTails.keys(),
+    ];
+  }
+
   documentSetSize(): number {
     return (this as unknown as { documentSet: Set<string> }).documentSet.size;
   }
@@ -442,9 +574,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 };
@@ -1318,6 +1451,611 @@ 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<void>((resolve) => {
+      releaseFirst = resolve;
+    });
+    let releaseSecond!: () => void;
+    const secondBlocked = new Promise<void>((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<void>((resolve) => {
+      releaseFirst = resolve;
+    });
+    let promptCount = 0;
+    (bridge.prompt as ReturnType<typeof vi.fn>).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('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<void>((resolve) => {
+      releaseFirst = resolve;
+    });
+    const secondBlocked = new Promise<void>((resolve) => {
+      releaseSecond = resolve;
+    });
+    let promptCount = 0;
+    (bridge.prompt as ReturnType<typeof vi.fn>).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: {
+      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(),
+      'classified-order-dws',
+      {
+        channelMemory: makeChannelMemory(),
+        memoryIntentClassifier,
+      },
+    );
+    (bridge.cancelSession as ReturnType<typeof vi.fn>).mockResolvedValue(
+      undefined,
+    );
+
+    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<typeof vi.fn>).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<void>((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('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<void>((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);
+    let release!: () => void;
+    const blocked = new Promise<void>((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('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<void>((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);
+    channel.seedPendingMessages(5_000);
+
+    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.',
+    );
+
+    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<void>((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<void>(() => 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('rejects admission instead of evicting a pending message', async () => {
+    const client = new FakeDwsClient();
+    const channel = await readyChannel(client);
+    channel.seedPendingMessages(5_000);
+
+    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.',
+    );
+
+    expect(channel.inbound).toEqual([]);
+    expect(channel.pendingMessageIds()).toHaveLength(5_000);
+    expect(channel.pendingMessageIds()[0]).toBe('parked-0');
+    expect(channel.pendingMessageIds()).not.toContain('new-request');
+  });
+
+  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()!;
+    channel.seedPendingMessages(5_000, true);
+    let release!: () => void;
+    const blocked = new Promise<void>((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 {
+      await channel.poll();
+
+      expect(channel.notificationWatermark()).toBe(initialWatermark);
+      expect(channel.pendingMessageCapacityWaiterCount()).toBe(0);
+
+      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('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);
+    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);
@@ -1828,6 +2566,91 @@ 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<void>((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));
+  });
+
+  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<void>((resolve) => {
+      releaseDocument = resolve;
+    });
+    let releaseOrdinary!: () => void;
+    const ordinaryBlocked = new Promise<void>((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
@@ -1849,10 +2672,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);
@@ -1870,9 +2694,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();
@@ -1916,31 +2740,38 @@ 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');
+      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();
 
-      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();
     }
   });
 
-  // 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();
@@ -2806,10 +3637,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',
@@ -2822,8 +3655,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 () => {
@@ -4535,8 +5370,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(
@@ -4680,6 +5515,139 @@ 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<void>((_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('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';
@@ -4843,7 +5811,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.
@@ -5067,6 +6035,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);
@@ -5087,8 +6061,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 6da936ea2db..dc5626984a9 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,
@@ -37,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.
  *
@@ -152,6 +154,11 @@ interface ImSubscriptionState {
   restartAttempts: number;
 }
 
+interface DirectConversationTail {
+  started: Promise<void>;
+  completed: Promise<void>;
+}
+
 interface ActiveReaction {
   target: { conversationId: string; messageId: string };
   sessionId: string;
@@ -518,6 +525,16 @@ export class DwsChannel extends PollingChannelBase<DwsCursor> {
   private readonly endReactionKeys = new Set<string>();
   private readonly notifiedSenderPairingNotifications = new Set<string>();
   private readonly processingMessages = new Map<string, Promise<void>>();
+  private readonly queuedDirectMessages = new Map<string, Promise<void>>();
+  // 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<string, Promise<void>>();
+  private readonly directConversationTails = new Map<
+    string,
+    DirectConversationTail
+  >();
+  private readonly directMessageStartResolvers = new Map<string, () => void>();
+  private readonly pendingMessageCapacityWaiters = new Set<() => void>();
   private pollAbortController = new AbortController();
   private lifecycleGeneration = 0;
   private connectionStartedAt = 0;
@@ -825,6 +842,13 @@ export class DwsChannel extends PollingChannelBase<DwsCursor> {
       this.cleanupReaction(key, 'disconnect reaction removal');
     }
     this.sessionReactionKeys.clear();
+    this.queuedDirectMessages.clear();
+    this.replayDirectDispatches.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();
     this.stopPollLoop();
     for (const state of this.imStates) {
       if (state.retryTimer) clearTimeout(state.retryTimer);
@@ -1124,7 +1148,7 @@ export class DwsChannel extends PollingChannelBase<DwsCursor> {
       );
       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 = {
@@ -1171,22 +1195,13 @@ export class DwsChannel extends PollingChannelBase<DwsCursor> {
         // `replayPendingMessages`; dispatching it here too would spend the
         // shared retry budget twice per poll.
         if (this.hasPendingMessage(key)) continue;
-        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.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 (signal.aborted || !this.connected) return;
       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
@@ -1370,7 +1385,9 @@ export class DwsChannel extends PollingChannelBase<DwsCursor> {
       (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,27 +1510,71 @@ export class DwsChannel extends PollingChannelBase<DwsCursor> {
     );
   }
 
-  private async handleImMessage(
-    source: DwsImSource,
+  private async receiveImMessage(
+    source: Exclude<DwsImSource, { kind: 'direct' }>,
     message: DwsImMessage,
     fromHistory = false,
   ): Promise<void> {
     if (!this.connected) return;
-    if (this.isSelfMessage(message)) {
+    if (this.markSelfMessageProcessed(message)) return;
+    if (
+      this.isStaleLiveMessage(message, fromHistory) &&
+      message.eventTime !== undefined
+    ) {
       this.markProcessedMessage(messageKey(message));
       this.saveCursor();
       return;
     }
     if (
-      !fromHistory &&
-      message.eventTime !== undefined &&
-      message.eventTime < this.connectionStartedAt - 5_000
+      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,
+    reportFailure = fromHistory,
+  ): DwsImDispatch {
+    const admission = this.admitReceivedDirectMessage(
+      message,
+      fromHistory,
+      reportFailure,
+    );
+    const completed = admission.then(({ completion }) => completion);
+    void completed.catch(() => undefined);
+    return {
+      admitted: admission.then(() => undefined),
+      completed,
+    };
+  }
+
+  private async admitReceivedDirectMessage(
+    message: DwsImMessage,
+    fromHistory: boolean,
+    reportFailure: boolean,
+  ): Promise<{ completion: Promise<void> }> {
+    if (!this.connected) return { completion: Promise.resolve() };
+    const key = messageKey(message);
+    if (this.markSelfMessageProcessed(message)) {
+      return { completion: Promise.resolve() };
+    }
+    if (
+      this.isStaleLiveMessage(message, fromHistory) &&
+      message.eventTime !== undefined
     ) {
-      if (source.kind !== 'direct') {
-        this.markProcessedMessage(messageKey(message));
-        this.saveCursor();
-        return;
-      }
       // 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,127 @@ export class DwsChannel extends PollingChannelBase<DwsCursor> {
         `[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;
+      return { completion: Promise.resolve() };
     }
-    if (
-      source.kind === 'group-all' &&
-      (this.config.groups[message.conversationId]?.requireMention ??
-        this.config.groups['*']?.requireMention ??
-        true)
-    ) {
-      return;
+    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<PersistedPendingMessage['source'], { kind: 'direct' }>,
+    message: DwsImMessage,
+    key: string,
+    reportFailure: boolean,
+  ): Promise<{ completion: Promise<void> }> {
+    if (this.cursor.processedMessages.includes(key)) {
+      this.removePersistedPendingMessage(key);
+      return { completion: Promise.resolve() };
+    }
+    if (!this.hasPendingMessage(key)) {
+      if (!(await this.rememberPendingMessage(source, message))) {
+        return { completion: Promise.resolve() };
+      }
     }
-    if (
-      (source.kind === 'group' || source.kind === 'group-all') &&
-      this.config.groupPolicy === 'pairing' &&
-      !this.groupGate.isGroupApproved(message.conversationId)
-    ) {
-      return;
+    return {
+      completion: this.scheduleDirectMessage(
+        source,
+        message,
+        key,
+        reportFailure,
+      ),
+    };
+  }
+
+  private scheduleDirectMessage(
+    source: Extract<PersistedPendingMessage['source'], { kind: 'direct' }>,
+    message: DwsImMessage,
+    key: string,
+    reportFailure: boolean,
+  ): Promise<void> {
+    const queued = this.queuedDirectMessages.get(key);
+    if (queued) return queued;
+    if (this.cursor.processedMessages.includes(key)) {
+      this.removePersistedPendingMessage(key);
+      return Promise.resolve();
     }
-    const key = messageKey(message);
+    const generation = this.lifecycleGeneration;
+    const dispatch = 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);
+        }
+      }
+    };
+    let resolveStarted!: () => void;
+    const started = new Promise<void>((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);
+    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(
+          source,
+          error instanceof Error ? error : new Error(String(error)),
+        );
+      });
+    }
+    return task;
+  }
+
+  private async dispatchImMessage(
+    source: DwsImSource,
+    message: DwsImMessage,
+    key: string,
+  ): Promise<void> {
     let waitedOnInFlight = false;
     let inFlightError: unknown;
     while (true) {
@@ -1580,6 +1744,12 @@ export class DwsChannel extends PollingChannelBase<DwsCursor> {
     this.processingMessages.set(key, task);
     try {
       await task;
+      if (
+        this.cursor.processedMessages.includes(key) &&
+        this.hasPendingMessage(key)
+      ) {
+        this.removePersistedPendingMessage(key);
+      }
     } finally {
       if (this.processingMessages.get(key) === task) {
         this.processingMessages.delete(key);
@@ -1651,7 +1821,7 @@ export class DwsChannel extends PollingChannelBase<DwsCursor> {
         // 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.rememberFailedMessage(source, message);
       }
       // Under budget the throw propagates exactly as before, so redelivery
       // and concurrent-duplicate retry keep their contracts. Once the budget
@@ -1738,30 +1908,72 @@ export class DwsChannel extends PollingChannelBase<DwsCursor> {
     );
   }
 
-  private rememberPendingMessage(
+  private async rememberPendingMessage(
     source: PersistedPendingMessage['source'],
     message: DwsImMessage,
-  ): void {
+  ): Promise<boolean> {
     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;
+    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 ?? [];
     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 {
-    this.cursor.pendingMessages = (this.cursor.pendingMessages ?? []).filter(
-      (pending) => messageKey(pending.message) !== key,
+  private async rememberFailedMessage(
+    source: PersistedPendingMessage['source'],
+    message: DwsImMessage,
+  ): Promise<void> {
+    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<void>((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(
+      (item) => messageKey(item.message) !== key,
     );
+    this.cursor.pendingMessages = remaining;
+    if (remaining.length === pending.length) return false;
+    for (const resolve of this.pendingMessageCapacityWaiters) resolve();
+    this.pendingMessageCapacityWaiters.clear();
+    return true;
   }
 
   private async processDocumentNotification(
@@ -1784,18 +1996,24 @@ export class DwsChannel extends PollingChannelBase<DwsCursor> {
       // 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 () => {
@@ -1901,8 +2119,29 @@ export class DwsChannel extends PollingChannelBase<DwsCursor> {
     for (const pending of [...(this.cursor.pendingMessages ?? [])]) {
       if (signal.aborted || !this.connected) return;
       const key = messageKey(pending.message);
+      if (pending.source.kind === 'direct') {
+        if (this.queuedDirectMessages.has(key)) continue;
+        if (this.replayDirectDispatches.size >= MAX_DIRECT_REPLAY_DISPATCHES) {
+          continue;
+        }
+        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 {
-        await this.handleImMessage(pending.source, pending.message, true);
+        await this.receiveImMessage(pending.source, pending.message, true);
         this.removePendingMessage(key);
         this.saveCursor();
       } catch (error) {
@@ -1961,9 +2200,14 @@ export class DwsChannel extends PollingChannelBase<DwsCursor> {
     );
   }
 
-  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),
     );
   }
 
@@ -2043,6 +2287,17 @@ export class DwsChannel extends PollingChannelBase<DwsCursor> {
     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,
@@ -2235,6 +2490,9 @@ export class DwsChannel extends PollingChannelBase<DwsCursor> {
 
   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;
     }
diff --git a/packages/channels/dws/src/dws-client.test.ts b/packages/channels/dws/src/dws-client.test.ts
index 6ce4840ff86..47ef615ba7f 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,205 @@ describe('DwsClient', () => {
     });
   });
 
+  it('does not make the direct event reader wait for message processing', async () => {
+    let onLine!: (line: string) => void | Promise<void>;
+    const eventStarter = vi.fn<DwsEventProcessStarter>(
+      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<void>((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<void>;
+    const eventStarter = vi.fn<DwsEventProcessStarter>(
+      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('does not report completion after direct admission fails', async () => {
+    let onLine!: (line: string) => void | Promise<void>;
+    const eventStarter = vi.fn<DwsEventProcessStarter>(
+      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<void>;
+    const eventStarter = vi.fn<DwsEventProcessStarter>(
+      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<void>((resolve) => {
+      releaseAdmission = resolve;
+    });
+    let rejectCompletion!: (error: Error) => void;
+    const completed = new Promise<void>((_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<DwsEventProcessStarter>(async () =>
       subscription(),
diff --git a/packages/channels/dws/src/dws-client.ts b/packages/channels/dws/src/dws-client.ts
index 4adb552ae78..1ef21c53d00 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<void>;
+  completed: Promise<void>;
+}
+
+export type DwsImMessageResult = void | Promise<void> | DwsImDispatch;
+
 export interface DwsTodoTask {
   taskId: string;
   title: string;
@@ -69,7 +76,7 @@ export interface DwsClientLike {
   assertAuthenticated(signal?: AbortSignal): Promise<DwsIdentity>;
   subscribeToIm(
     source: DwsImSource,
-    onMessage: (message: DwsImMessage) => void | Promise<void>,
+    onMessage: (message: DwsImMessage) => DwsImMessageResult,
     onError: (error: Error) => void,
   ): Promise<DwsEventSubscription>;
   sendImMessage(
@@ -750,7 +757,7 @@ export class DwsClient implements DwsClientLike {
 
   async subscribeToIm(
     source: DwsImSource,
-    onMessage: (message: DwsImMessage) => void | Promise<void>,
+    onMessage: (message: DwsImMessage) => DwsImMessageResult,
     onError: (error: Error) => void,
   ): Promise<DwsEventSubscription> {
     const args = [
@@ -767,7 +774,27 @@ 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;
+          }
+        };
+        const admissionSucceeded = result.admitted.then(
+          () => true,
+          () => false,
+        );
+        void result.completed.catch(async (error: unknown) => {
+          if (await admissionSucceeded) reportError(error);
+        });
+        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,