fix(channels): Make same-chat delivery session-aware (#10145)

Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com>

Author: doudouOUC

docs/design/channel-session-aware-delivery.md

@@ -0,0 +1,78 @@
+# Channel Session-Aware Delivery
+
+## Problem
+
+Channel sessions are already independent, but a few adapters keep delivery-only
+state at chat or process scope. When two sessions run in the same chat, a later
+message can replace the reply target of an earlier QQ response, one completed
+Weixin session can clear another session's typing indicator, and the plugin
+example can associate overlapping responses with the wrong inbound message.
+
+## Scope
+
+This change makes existing same-chat delivery state session-aware. It does not
+add named sessions, task selection, session persistence, result labels, or
+worktree creation. It does not change `ChannelBase` or the public channel
+adapter interface.
+
+`ChannelOutputSegmentContext` already carries the originating session, run,
+target, and message ID. `ChannelBase` also retains the active prompt's message
+ID until response delivery completes. Adapters can therefore preserve the
+origin without introducing another shared delivery abstraction.
+
+## QQ
+
+QQ passive replies require the original inbound `msg_id`, and streaming blocks
+for that message share a `msg_seq` counter. The adapter keeps a bounded
+in-memory context for every recently accepted message while retaining the
+persisted per-chat latest message as a compatibility pointer.
+
+Prompt output uses the message ID from the active prompt or output segment.
+Streaming state copies that reply context so delayed flushes and retries do not
+depend on mutable chat state. Replies produced while handling an inbound
+command use async-local inbound context. Background and cron delivery is
+explicitly active and never borrows the latest inbound message.
+
+An expired or missing explicit context falls back to active delivery, never to
+a different message in the same chat. Sequence counters remain isolated by
+message ID and are reclaimed with the message context's existing five-minute
+TTL. The persisted schema is unchanged; restore keeps sequence counters only
+for restored valid reply contexts.
+
+## Weixin
+
+Weixin typing ownership is tracked as `chatId -> sessionId -> startedAt`.
+Starting the first session enables typing and starts one chat-level keepalive.
+Additional sessions only add owners. A terminal event removes its own owner,
+and only the last owner disables typing.
+
+The existing generation guard continues to reject stale asynchronous typing
+results. The keepalive backstop expires individual sessions rather than the
+whole chat, so an old wedged session cannot clear a newer session's indicator.
+Session death removes only that session; disconnect clears all state.
+
+## Plugin example
+
+The example replaces its process-global pending message ID with async-local
+inbound context for command replies and uses output-segment or active-prompt
+message IDs for agent output. This demonstrates the same correlation contract
+to third-party adapter authors without changing the protocol.
+
+## Failure semantics
+
+- A missing or expired QQ passive context uses active delivery and remains
+  subject to the existing active-message policy.
+- A failed QQ passive attempt rolls back only its original message's sequence
+  before the existing active fallback.
+- A failed Weixin typing request does not discard live owners; the bounded
+  keepalive retries while at least one owner remains.
+- Cleanup on session death, group removal, and disconnect cannot remove state
+  owned by another live session.
+
+## Verification
+
+Focused tests cover two sessions in one chat, delayed streaming flushes,
+independent QQ sequence rollback, expired reply contexts, background delivery,
+Weixin first-owner/last-owner transitions, stale async typing results,
+per-session backstop expiry, and overlapping plugin messages. Package tests are
+followed by the repository build, typecheck, and lint checks.

packages/channels/plugin-example/package.json

@@ -21,6 +21,7 @@
   },
   "scripts": {
     "build": "tsc --build",
+    "test": "vitest run",
     "prepublishOnly": "npm run build"
   },
   "dependencies": {

packages/channels/plugin-example/src/MockPluginChannel.test.ts

@@ -0,0 +1,161 @@
+import { EventEmitter } from 'node:events';
+import { beforeEach, describe, expect, it, vi } from 'vitest';
+import type { Envelope } from '@qwen-code/channel-base';
+
+const baseHandleInbound = vi.hoisted(() => vi.fn());
+
+vi.mock('@qwen-code/channel-base', () => ({
+  ChannelBase: class {
+    protected name: string;
+
+    constructor(name: string) {
+      this.name = name;
+    }
+
+    protected handleInbound(envelope: Envelope): Promise<void> {
+      return baseHandleInbound.call(this, envelope) as Promise<void>;
+    }
+
+    protected getResponseMessageId(): string | undefined {
+      return undefined;
+    }
+  },
+}));
+
+vi.mock('ws', () => ({
+  default: class MockWebSocket {
+    static OPEN = 1;
+  },
+}));
+
+import { MockPluginChannel } from './MockPluginChannel.js';
+
+function deferredPromise() {
+  let resolve!: () => void;
+  const promise = new Promise<void>((res) => {
+    resolve = res;
+  });
+  return { promise, resolve };
+}
+
+function envelope(messageId: string): Envelope {
+  return {
+    channelName: 'mock',
+    senderId: `sender-${messageId}`,
+    senderName: messageId,
+    chatId: 'shared-chat',
+    text: messageId,
+    messageId,
+    isGroup: false,
+    isMentioned: false,
+    isReplyToBot: false,
+  };
+}
+
+function createChannel() {
+  const channel = new MockPluginChannel(
+    'mock',
+    {
+      type: 'mock',
+      token: 'test-token',
+      serverWsUrl: 'ws://example.test',
+      senderPolicy: 'open',
+      allowedUsers: [],
+      sessionScope: 'user',
+      cwd: process.cwd(),
+      groupPolicy: 'disabled',
+      dmPolicy: 'open',
+      groups: {},
+    },
+    new EventEmitter() as never,
+  );
+  const send = vi.fn();
+  (channel as unknown as { ws: { readyState: number; send: typeof send } }).ws =
+    { readyState: 1, send };
+  return { channel, send };
+}
+
+describe('MockPluginChannel message correlation', () => {
+  beforeEach(() => {
+    baseHandleInbound.mockReset();
+  });
+
+  it('keeps overlapping inbound command replies bound to their own messages', async () => {
+    const { channel, send } = createChannel();
+    const first = deferredPromise();
+    const second = deferredPromise();
+    baseHandleInbound.mockImplementation(async function (
+      this: MockPluginChannel,
+      inbound: Envelope,
+    ) {
+      await (inbound.messageId === 'msg-a' ? first.promise : second.promise);
+      await this.sendMessage(inbound.chatId, `reply-${inbound.messageId}`);
+    });
+
+    const pendingA = channel.handleInbound(envelope('msg-a'));
+    const pendingB = channel.handleInbound(envelope('msg-b'));
+    second.resolve();
+    await pendingB;
+    first.resolve();
+    await pendingA;
+
+    expect(send.mock.calls.map(([frame]) => JSON.parse(String(frame)))).toEqual(
+      [
+        {
+          type: 'outbound',
+          messageId: 'msg-b',
+          chatId: 'shared-chat',
+          text: 'reply-msg-b',
+        },
+        {
+          type: 'outbound',
+          messageId: 'msg-a',
+          chatId: 'shared-chat',
+          text: 'reply-msg-a',
+        },
+      ],
+    );
+  });
+
+  it('uses output segment message IDs for chunks and final responses', async () => {
+    const { channel, send } = createChannel();
+    const output = channel as unknown as {
+      onResponseChunk: (
+        chatId: string,
+        chunk: string,
+        sessionId: string,
+        segment: { messageId: string },
+      ) => void;
+      onResponseComplete: (
+        chatId: string,
+        text: string,
+        sessionId: string,
+        segment: { messageId: string },
+      ) => Promise<void>;
+    };
+
+    output.onResponseChunk('shared-chat', 'chunk-a', 'session-a', {
+      messageId: 'msg-a',
+    });
+    await output.onResponseComplete('shared-chat', 'final-b', 'session-b', {
+      messageId: 'msg-b',
+    });
+
+    expect(send.mock.calls.map(([frame]) => JSON.parse(String(frame)))).toEqual(
+      [
+        {
+          type: 'chunk',
+          messageId: 'msg-a',
+          chatId: 'shared-chat',
+          text: 'chunk-a',
+        },
+        {
+          type: 'outbound',
+          messageId: 'msg-b',
+          chatId: 'shared-chat',
+          text: 'final-b',
+        },
+      ],
+    );
+  });
+});

packages/channels/plugin-example/src/MockPluginChannel.ts

@@ -4,8 +4,10 @@ import type {
   ChannelBaseOptions,
   Envelope,
   ChannelAgentBridge,
+  ChannelOutputSegmentContext,
 } from '@qwen-code/channel-base';
 import WebSocket from 'ws';
+import { AsyncLocalStorage } from 'node:async_hooks';
 import type {
   InboundMessage,
   OutboundMessage,
@@ -19,7 +21,7 @@ export interface MockPluginConfig extends ChannelConfig {
 export class MockPluginChannel extends ChannelBase {
   private ws: WebSocket | null = null;
   private serverWsUrl: string;
-  private pendingMessageId: string | undefined;
+  private inboundMessage = new AsyncLocalStorage<{ messageId?: string }>();
 
   constructor(
     name: string,
@@ -83,13 +85,15 @@ export class MockPluginChannel extends ChannelBase {
   protected override onResponseChunk(
     chatId: string,
     chunk: string,
-    _sessionId: string,
+    sessionId: string,
+    segment?: ChannelOutputSegmentContext,
   ): void {
     if (!this.ws || this.ws.readyState !== WebSocket.OPEN) return;
 
     const msg: ChunkMessage = {
       type: 'chunk',
-      messageId: this.pendingMessageId || 'unknown',
+      messageId:
+        segment?.messageId ?? this.getResponseMessageId(sessionId) ?? 'unknown',
       chatId,
       text: chunk,
     };
@@ -99,19 +103,28 @@ export class MockPluginChannel extends ChannelBase {
   protected override async onResponseComplete(
     chatId: string,
     fullText: string,
-    _sessionId: string,
+    sessionId: string,
+    segment?: ChannelOutputSegmentContext,
   ): Promise<void> {
-    await this.sendMessage(chatId, fullText);
+    this.sendOutbound(
+      chatId,
+      fullText,
+      segment?.messageId ?? this.getResponseMessageId(sessionId),
+    );
   }
 
   async sendMessage(chatId: string, text: string): Promise<void> {
+    this.sendOutbound(chatId, text, this.inboundMessage.getStore()?.messageId);
+  }
+
+  private sendOutbound(chatId: string, text: string, messageId?: string): void {
     if (!this.ws || this.ws.readyState !== WebSocket.OPEN) {
       return;
     }
 
     const outbound: OutboundMessage = {
       type: 'outbound',
-      messageId: this.pendingMessageId || 'unknown',
+      messageId: messageId ?? 'unknown',
       chatId,
       text,
     };
@@ -127,11 +140,8 @@ export class MockPluginChannel extends ChannelBase {
   }
 
   override async handleInbound(envelope: Envelope): Promise<void> {
-    this.pendingMessageId = envelope.messageId;
-    try {
-      await super.handleInbound(envelope);
-    } finally {
-      this.pendingMessageId = undefined;
-    }
+    await this.inboundMessage.run({ messageId: envelope.messageId }, async () =>
+      super.handleInbound(envelope),
+    );
   }
 }

packages/channels/plugin-example/tsconfig.json

@@ -5,6 +5,6 @@
     "rootDir": "src"
   },
   "include": ["src/**/*.ts"],
-  "exclude": ["node_modules", "dist"],
+  "exclude": ["node_modules", "dist", "src/**/*.test.ts"],
   "references": [{ "path": "../base" }]
 }

packages/channels/plugin-example/vitest.config.ts

@@ -0,0 +1,17 @@
+import path from 'node:path';
+import { defineConfig } from 'vitest/config';
+
+export default defineConfig({
+  test: {
+    include: ['src/**/*.test.ts'],
+    globals: true,
+  },
+  resolve: {
+    alias: {
+      '@qwen-code/channel-base': path.resolve(
+        __dirname,
+        '../base/src/index.ts',
+      ),
+    },
+  },
+});