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

Co-authored-by: Qwen-Coder <[email protected]>

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',
+      ),
+    },
+  },
+});

packages/channels/qqbot/src/QQChannel.ts

@@ -25,9 +25,12 @@ import type {
   ChannelConfig,
   ChannelBaseOptions,
   ChannelAgentBridge,
+  ChannelOutputSegmentContext,
+  Envelope,
   ToolCallEvent,
 } from '@qwen-code/channel-base';
 import WebSocket from 'ws';
+import { AsyncLocalStorage } from 'node:async_hooks';
 import {
   readFileSync,
   writeFileSync,
@@ -78,6 +81,20 @@ export class DeliveryError extends Error {
   }
 }
 
+interface QQReplyContext {
+  chatId: string;
+  msgId: string;
+  timestamp: number;
+}
+
+interface QQStreamState {
+  chatId: string;
+  buffer: string;
+  timer: ReturnType<typeof setTimeout> | null;
+  retryCount: number;
+  replyContext?: QQReplyContext;
+}
+
 /** Validate chatId to prevent SSRF when constructing URLs. */
 export function isValidChatId(id: string): boolean {
   return /^[A-Za-z0-9_-]+$/.test(id) && id.length <= 128;
@@ -126,6 +143,8 @@ export class QQChannel extends ChannelBase {
   /** Track the latest user messageId per chatId for proper reply (msg_id). */
   private replyMsgId: Map<string, { msgId: string; timestamp: number }> =
     new Map();
+  private replyContextByMessageId = new Map<string, QQReplyContext>();
+  private inboundReplyContext = new AsyncLocalStorage<QQReplyContext>();
   /** msg_seq counter per user messageId, for multi-block streaming. */
   private msgSeqMap: Map<string, number> = new Map();
   /** Periodic cleanup timer for expired replyMsgId entries. */
@@ -200,15 +219,7 @@ export class QQChannel extends ChannelBase {
    *   - flushedSessions tracks already-sent sessions to skip final fullText
    */
   // ── Streaming state ───────────────────────────────────────────
-  private streamState: Map<
-    string,
-    {
-      chatId: string;
-      buffer: string;
-      timer: ReturnType<typeof setTimeout> | null;
-      retryCount: number;
-    }
-  > = new Map();
+  private streamState = new Map<string, QQStreamState>();
   private flushingSessions: Set<string> = new Set();
   private pendingStreamDelete: Set<string> = new Set();
   private _reconnectId: number = 0;
@@ -348,7 +359,7 @@ export class QQChannel extends ChannelBase {
         if (toFlush) {
           const target = this.router.getTarget(sessionId);
           if (target) {
-            this.sendMessage(target.chatId, toFlush)
+            this.sendMessageWithReplyContext(target.chatId, toFlush)
               .then(() => {
                 if (!entry!.buffer && this.cronBuffer.get(sessionId) === entry)
                   this.cronBuffer.delete(sessionId);
@@ -388,7 +399,7 @@ export class QQChannel extends ChannelBase {
                     this.cronBuffer.delete(sessionId);
                     return;
                   }
-                  this.sendMessage(retryTarget.chatId, toFlush)
+                  this.sendMessageWithReplyContext(retryTarget.chatId, toFlush)
                     .then(() => {
                       entry!.pendingRetry = '';
                       if (
@@ -441,7 +452,10 @@ export class QQChannel extends ChannelBase {
                             this.cronBuffer.delete(sessionId);
                             return;
                           }
-                          this.sendMessage(retryTarget2.chatId, toFlush)
+                          this.sendMessageWithReplyContext(
+                            retryTarget2.chatId,
+                            toFlush,
+                          )
                             .then(() => {
                               entry!.pendingRetry = '';
                               if (
@@ -620,7 +634,48 @@ export class QQChannel extends ChannelBase {
     }
   }
 
+  override async handleInbound(envelope: Envelope): Promise<void> {
+    const context = envelope.messageId
+      ? this.replyContextByMessageId.get(envelope.messageId)
+      : undefined;
+    if (!context || context.chatId !== envelope.chatId) {
+      await super.handleInbound(envelope);
+      return;
+    }
+    await this.inboundReplyContext.run(context, () =>
+      super.handleInbound(envelope),
+    );
+  }
+
   async sendMessage(chatId: string, text: string): Promise<void> {
+    const inboundContext = this.inboundReplyContext.getStore();
+    const latest = this.replyMsgId.get(chatId);
+    const replyContext =
+      inboundContext?.chatId === chatId
+        ? inboundContext
+        : latest
+          ? { chatId, ...latest }
+          : undefined;
+    await this.sendMessageWithReplyContext(chatId, text, replyContext);
+  }
+
+  protected override async sendResponseMessage(
+    chatId: string,
+    text: string,
+    sessionId: string,
+  ): Promise<void> {
+    const messageId = this.getResponseMessageId(sessionId);
+    const replyContext = messageId
+      ? this.replyContextByMessageId.get(messageId)
+      : undefined;
+    await this.sendMessageWithReplyContext(chatId, text, replyContext);
+  }
+
+  private async sendMessageWithReplyContext(
+    chatId: string,
+    text: string,
+    replyContext?: QQReplyContext,
+  ): Promise<void> {
     // <noreply> suppression
     if (text.trim() === '<noreply>') {
       process.stderr.write(
@@ -632,8 +687,7 @@ export class QQChannel extends ChannelBase {
     const route = await this.resolveRoute(chatId);
     if (!route) return;
 
-    // Look up reply context with TTL check
-    const entry = this.replyMsgId.get(chatId);
+    const entry = replyContext?.chatId === chatId ? replyContext : undefined;
     const msgId =
       entry && Date.now() - entry.timestamp < QQChannel.REPLY_MSG_ID_TTL_MS
         ? entry.msgId
@@ -642,8 +696,7 @@ export class QQChannel extends ChannelBase {
       process.stderr.write(
         `[QQ:${this.name}] replyMsgId entry expired for ${sanitizeLogText(chatId, 64)}, reply context expired, sending without msg_id\n`,
       );
-      this.msgSeqMap.delete(entry.msgId);
-      this.replyMsgId.delete(chatId);
+      this.deleteReplyContext(entry);
       this.saveQQState();
     }
 
@@ -970,6 +1023,7 @@ export class QQChannel extends ChannelBase {
     this.detachCronHandler();
     this.chatTypeMap.clear();
     this.replyMsgId.clear();
+    this.replyContextByMessageId.clear();
     this.msgSeqMap.clear();
     this.botOpenIdByGroup.clear();
     this.warnedSenderOpenIds.clear();
@@ -1021,15 +1075,22 @@ export class QQChannel extends ChannelBase {
     chatId: string,
     chunk: string,
     sessionId: string,
+    segment?: ChannelOutputSegmentContext,
   ): void {
     if (this.blockStreaming) return;
     let state = this.streamState.get(sessionId);
     if (!state) {
-      state = { chatId, buffer: chunk, timer: null, retryCount: 0 } as {
-        chatId: string;
-        buffer: string;
-        timer: ReturnType<typeof setTimeout> | null;
-        retryCount: number;
+      const messageId =
+        segment?.messageId ?? this.getResponseMessageId(sessionId);
+      const replyContext = messageId
+        ? this.replyContextByMessageId.get(messageId)
+        : undefined;
+      state = {
+        chatId,
+        buffer: chunk,
+        timer: null,
+        retryCount: 0,
+        ...(replyContext ? { replyContext } : {}),
       };
       this.streamState.set(sessionId, state);
     } else {
@@ -1099,19 +1160,14 @@ export class QQChannel extends ChannelBase {
   private flushAndTrack(
     sessionId: string,
     buffer: string,
-    state: {
-      chatId: string;
-      buffer: string;
-      timer: ReturnType<typeof setTimeout> | null;
-      retryCount: number;
-    },
+    state: QQStreamState,
     logLabel: string,
   ): void {
     this.flushingSessions.add(sessionId);
     // sendMessage throws DeliveryError for delivery failures.
     // RETRY_EXHAUSTED, ACTIVE_MSG_DISABLED, and FALLBACK_FAILED are
     // permanent. RATE_LIMITED is transient and falls through to re-buffer/retry.
-    this.sendMessage(state.chatId, buffer)
+    this.sendMessageWithReplyContext(state.chatId, buffer, state.replyContext)
       .then(() => {
         // #3: Guard — if session died during in-flight send, touch nothing
         const current = this.streamState.get(sessionId);
@@ -1479,6 +1535,8 @@ export class QQChannel extends ChannelBase {
                     (o['msgId'] as string).length <= 128 &&
                     typeof o['timestamp'] === 'number' &&
                     Number.isFinite(o['timestamp']) &&
+                    o['timestamp'] >=
+                      Date.now() - QQChannel.REPLY_MSG_ID_TTL_MS &&
                     o['timestamp'] <= Date.now() + QQChannel.REPLY_MSG_ID_TTL_MS
                   );
                 })
@@ -1496,6 +1554,12 @@ export class QQChannel extends ChannelBase {
           );
         }
       }
+      this.replyContextByMessageId = new Map(
+        Array.from(this.replyMsgId, ([chatId, entry]) => [
+          entry.msgId,
+          { chatId, ...entry },
+        ]),
+      );
       if (raw.msgSeqMap) {
         const arr = raw.msgSeqMap as Array<[string, unknown]>;
         const totalRaw = Array.isArray(arr) ? arr.length : 0;
@@ -1517,6 +1581,11 @@ export class QQChannel extends ChannelBase {
           );
         }
       }
+      for (const msgId of this.msgSeqMap.keys()) {
+        if (!this.replyContextByMessageId.has(msgId)) {
+          this.msgSeqMap.delete(msgId);
+        }
+      }
       if (raw.groupActiveMsgEnabled) {
         const arr = raw.groupActiveMsgEnabled as Array<[string, unknown]>;
         const totalRaw = Array.isArray(arr) ? arr.length : 0;
@@ -1652,19 +1721,21 @@ export class QQChannel extends ChannelBase {
 
   // ── ReplyMsgId helpers ────────────────────────────────────────
 
-  /**
-   * Set replyMsgId for a chat, cleaning up the previous entry's msgSeqMap
-   * to prevent orphaned entries accumulating over time.
-   */
   private setReplyMsgId(chatId: string, msgId: string): void {
-    const oldEntry = this.replyMsgId.get(chatId);
-    if (oldEntry && oldEntry.msgId !== msgId) {
-      this.msgSeqMap.delete(oldEntry.msgId);
-    }
-    this.replyMsgId.set(chatId, { msgId, timestamp: Date.now() });
+    const timestamp = Date.now();
+    this.replyMsgId.set(chatId, { msgId, timestamp });
+    this.replyContextByMessageId.set(msgId, { chatId, msgId, timestamp });
     this.saveQQState();
   }
 
+  private deleteReplyContext(context: QQReplyContext): void {
+    this.replyContextByMessageId.delete(context.msgId);
+    this.msgSeqMap.delete(context.msgId);
+    if (this.replyMsgId.get(context.chatId)?.msgId === context.msgId) {
+      this.replyMsgId.delete(context.chatId);
+    }
+  }
+
   /**
    * Start periodic cleanup of expired replyMsgId entries.
    * Evicts entries older than 5 minutes every 60 seconds, and cascades
@@ -1675,6 +1746,12 @@ export class QQChannel extends ChannelBase {
     this.replyMsgIdCleanupTimer = setInterval(() => {
       const cutoff = Date.now() - QQChannel.REPLY_MSG_ID_TTL_MS;
       let dirty = false;
+      for (const context of this.replyContextByMessageId.values()) {
+        if (context.timestamp < cutoff) {
+          this.deleteReplyContext(context);
+          dirty = true;
+        }
+      }
       for (const [chatId, entry] of this.replyMsgId) {
         if (entry.timestamp < cutoff) {
           this.msgSeqMap.delete(entry.msgId);
@@ -2796,6 +2873,12 @@ export class QQChannel extends ChannelBase {
     const replyEntry = this.replyMsgId.get(groupId);
     if (replyEntry) this.msgSeqMap.delete(replyEntry.msgId);
     this.replyMsgId.delete(groupId);
+    for (const context of this.replyContextByMessageId.values()) {
+      if (context.chatId === groupId) {
+        this.replyContextByMessageId.delete(context.msgId);
+        this.msgSeqMap.delete(context.msgId);
+      }
+    }
     this.botOpenIdByGroup.delete(groupId);
     this._lastKeywordNoMatchLog.delete(groupId);
     // Clean up cron buffers targeting this group (always, regardless of config flag)

packages/channels/qqbot/src/cron.test.ts

@@ -54,6 +54,9 @@ vi.mock('@qwen-code/channel-base', () => ({
     protected handleInbound(_env: unknown): Promise<void> {
       return Promise.resolve();
     }
+    protected getResponseMessageId(_sessionId: string): string | undefined {
+      return undefined;
+    }
     protected onSessionDied(_sessionId: string): void {}
   },
   SessionRouter: class {

packages/channels/qqbot/src/persistence.test.ts

@@ -289,6 +289,7 @@ describe('restoreQQState', () => {
 
   it('restores msgSeqMap from disk', () => {
     fsStore[statePath] = JSON.stringify({
+      replyMsgId: [['u1', 'msg_abc']],
       msgSeqMap: [['msg_abc', 5]],
     });
     const ch = makeChannel();
@@ -299,6 +300,23 @@ describe('restoreQQState', () => {
     expect(msgSeqMap.get('msg_abc')).toBe(5);
   });
 
+  it('discards sequence counters without a restored reply context', () => {
+    fsStore[statePath] = JSON.stringify({
+      replyMsgId: [['u1', 'msg-current']],
+      msgSeqMap: [
+        ['msg-current', 2],
+        ['msg-orphan', 7],
+      ],
+    });
+    const ch = makeChannel();
+    (ch as unknown as { restoreQQState: () => boolean }).restoreQQState();
+
+    const msgSeqMap = (ch as unknown as { msgSeqMap: Map<string, number> })
+      .msgSeqMap;
+    expect(msgSeqMap.get('msg-current')).toBe(2);
+    expect(msgSeqMap.has('msg-orphan')).toBe(false);
+  });
+
   it('restores groupActiveMsgEnabled from disk', () => {
     fsStore[statePath] = JSON.stringify({
       groupActiveMsgEnabled: [
@@ -362,6 +380,7 @@ describe('restoreQQState', () => {
 
   it('filters invalid msgSeqMap values (negative or non-number)', () => {
     fsStore[statePath] = JSON.stringify({
+      replyMsgId: [['u1', 'a']],
       msgSeqMap: [
         ['a', 3],
         ['b', -1],

packages/channels/qqbot/src/send.test.ts

@@ -3,6 +3,7 @@ import { join } from 'node:path';
 import type {
   ChannelAgentBridge,
   ChannelTaskLifecycleEvent,
+  Envelope,
 } from '@qwen-code/channel-base';
 import { isValidChatId, DeliveryError } from './QQChannel.js';
 
@@ -12,6 +13,7 @@ const {
   mockFetchGatewayUrl,
   MockWebSocket,
   mockWebSockets,
+  mockBaseHandleInbound,
 } = vi.hoisted(() => {
   const mockWebSockets: unknown[] = [];
 
@@ -47,6 +49,7 @@ const {
     mockSendQQMessage: vi.fn(),
     mockFetchAccessToken: vi.fn(),
     mockFetchGatewayUrl: vi.fn(),
+    mockBaseHandleInbound: vi.fn(() => Promise.resolve()),
     MockWebSocket,
     mockWebSockets,
   };
@@ -105,8 +108,11 @@ vi.mock('@qwen-code/channel-base', async () => {
         this.router = (options?.['router'] as Record<string, unknown>) ?? {};
         this.baseOptions = options ?? ({} as Record<string, unknown>);
       }
-      protected handleInbound(_env: unknown): Promise<void> {
-        return Promise.resolve();
+      protected handleInbound(env: unknown): Promise<void> {
+        return mockBaseHandleInbound(env) as Promise<void>;
+      }
+      protected getResponseMessageId(_sessionId: string): string | undefined {
+        return undefined;
       }
       protected onTaskLifecycle(_event: unknown): void {}
     },
@@ -141,6 +147,14 @@ function mockResponse(
   return { ok, status, text: async () => body };
 }
 
+function deferredPromise() {
+  let resolve!: () => void;
+  const promise = new Promise<void>((res) => {
+    resolve = res;
+  });
+  return { promise, resolve };
+}
+
 describe('isValidChatId', () => {
   it('accepts alphanumeric IDs', () => {
     expect(isValidChatId('abc123')).toBe(true);
@@ -839,6 +853,108 @@ describe('sendMessage', () => {
     saveSpy.mockRestore();
   });
 
+  it('does not borrow the latest reply context for a background response', async () => {
+    const ch = makeChannel({ chatType: 'c2c', replyMsgId: 'msg-latest' });
+    const channel = ch as unknown as {
+      sendResponseMessage: (
+        chatId: string,
+        text: string,
+        sessionId: string,
+      ) => Promise<void>;
+    };
+
+    await channel.sendResponseMessage(
+      'test-chat-id',
+      'background',
+      'session-background',
+    );
+
+    expect(mockSendQQMessage.mock.calls[0][3]).toEqual({
+      msg_type: 2,
+      markdown: { content: 'background' },
+    });
+  });
+
+  it('keeps overlapping inbound replies bound to their own messages', async () => {
+    const ch = makeChannel({ chatType: 'c2c' });
+    const channel = ch as unknown as {
+      setReplyMsgId: (chatId: string, messageId: string) => void;
+    };
+    channel.setReplyMsgId('test-chat-id', 'msg-a');
+    channel.setReplyMsgId('test-chat-id', 'msg-b');
+    const first = deferredPromise();
+    const second = deferredPromise();
+    mockBaseHandleInbound
+      .mockImplementationOnce(async (inbound: Envelope) => {
+        await first.promise;
+        await ch.sendMessage(inbound.chatId, 'reply-a');
+      })
+      .mockImplementationOnce(async (inbound: Envelope) => {
+        await second.promise;
+        await ch.sendMessage(inbound.chatId, 'reply-b');
+      });
+    const base = {
+      channelName: 'test-bot',
+      senderId: 'sender',
+      senderName: 'sender',
+      chatId: 'test-chat-id',
+      text: 'prompt',
+      isGroup: false,
+      isMentioned: false,
+      isReplyToBot: false,
+    };
+
+    const pendingA = ch.handleInbound({ ...base, messageId: 'msg-a' });
+    const pendingB = ch.handleInbound({ ...base, messageId: 'msg-b' });
+    second.resolve();
+    await pendingB;
+    first.resolve();
+    await pendingA;
+
+    expect(mockSendQQMessage.mock.calls[0][3]).toMatchObject({
+      msg_id: 'msg-b',
+      msg_seq: 1,
+    });
+    expect(mockSendQQMessage.mock.calls[1][3]).toMatchObject({
+      msg_id: 'msg-a',
+      msg_seq: 1,
+    });
+  });
+
+  it('does not replace an expired session reply with a newer chat reply', async () => {
+    const ch = makeChannel({ chatType: 'c2c', replyMsgId: 'msg-newer' });
+    const channel = ch as unknown as {
+      getResponseMessageId: (sessionId: string) => string | undefined;
+      replyContextByMessageId: Map<
+        string,
+        { chatId: string; msgId: string; timestamp: number }
+      >;
+      sendResponseMessage: (
+        chatId: string,
+        text: string,
+        sessionId: string,
+      ) => Promise<void>;
+    };
+    channel.getResponseMessageId = () => 'msg-expired';
+    channel.replyContextByMessageId.set('msg-expired', {
+      chatId: 'test-chat-id',
+      msgId: 'msg-expired',
+      timestamp: Date.now() - 300_001,
+    });
+
+    await channel.sendResponseMessage(
+      'test-chat-id',
+      'late response',
+      'session-old',
+    );
+
+    expect(mockSendQQMessage.mock.calls[0][3]).toEqual({
+      msg_type: 2,
+      markdown: { content: 'late response' },
+    });
+    expect(channel.replyContextByMessageId.has('msg-expired')).toBe(false);
+  });
+
   it('sends single request even for long text (no splitting)', async () => {
     const ch = makeChannel({ chatType: 'c2c', replyMsgId: 'msg-789' });
     const text = 'a'.repeat(4500);
@@ -1160,7 +1276,7 @@ describe('setReplyMsgId', () => {
     return ch;
   }
 
-  it('cleans up old msgSeqMap entry when setting new replyMsgId for same chatId', () => {
+  it('retains the previous message sequence until its reply context expires', () => {
     const ch = makeChannel();
     const chp = ch as unknown as Record<string, unknown>;
 
@@ -1182,9 +1298,14 @@ describe('setReplyMsgId', () => {
       'new-msg-id',
     );
 
-    expect(msgSeqMap.has('old-msg-id')).toBe(false);
+    expect(msgSeqMap.get('old-msg-id')).toBe(5);
     expect(msgSeqMap.get('other-msg-id')).toBe(10);
     expect(replyMsgId.get('test-chat-id')!.msgId).toBe('new-msg-id');
+    const replyContexts = chp['replyContextByMessageId'] as Map<
+      string,
+      { chatId: string; msgId: string; timestamp: number }
+    >;
+    expect(replyContexts.get('new-msg-id')?.chatId).toBe('test-chat-id');
   });
 
   it('does nothing when chatId has no prior replyMsgId', () => {
@@ -1556,6 +1677,10 @@ describe('restoreQQState validation filters', () => {
     vi.mocked(existsSync).mockReturnValue(true);
     vi.mocked(readFileSync).mockReturnValue(
       JSON.stringify({
+        replyMsgId: [
+          ['chat-a', 'a'],
+          ['chat-b', 'b'],
+        ],
         msgSeqMap: [
           ['a', 0],
           ['b', 42],
@@ -1590,7 +1715,7 @@ describe('restoreQQState validation filters', () => {
     // Use raw JSON string: JSON.stringify(Infinity) → "null", which
     // bypasses Number.isSafeInteger (caught by typeof check instead).
     vi.mocked(readFileSync).mockReturnValue(
-      '{"msgSeqMap":[["a",1.5],["b",9007199254740992],["c",1e999],["d",-1e999],["e",42],["f",0]]}',
+      '{"replyMsgId":[["chat-e","e"],["chat-f","f"]],"msgSeqMap":[["a",1.5],["b",9007199254740992],["c",1e999],["d",-1e999],["e",42],["f",0]]}',
     );
 
     const ch = makeChannel();
@@ -1852,6 +1977,36 @@ describe('replyMsgId cleanup timer', () => {
     ch.disconnect();
   });
 
+  it('reclaims an older same-chat context without deleting the latest one', () => {
+    vi.useFakeTimers();
+    vi.setSystemTime(new Date('2026-01-01T00:00:00Z'));
+    const ch = makeChannel();
+    const chp = ch as unknown as Record<string, unknown>;
+    const setReplyMsgId = chp['setReplyMsgId'] as (
+      chatId: string,
+      msgId: string,
+    ) => void;
+    setReplyMsgId.call(ch, 'shared-chat', 'msg-old');
+    const msgSeqMap = chp['msgSeqMap'] as Map<string, number>;
+    msgSeqMap.set('msg-old', 3);
+
+    vi.advanceTimersByTime(60_000);
+    setReplyMsgId.call(ch, 'shared-chat', 'msg-latest');
+    msgSeqMap.set('msg-latest', 1);
+    (chp['startReplyMsgIdCleanup'] as () => void).call(ch);
+    vi.advanceTimersByTime(300_000);
+
+    const replyContexts = chp['replyContextByMessageId'] as Map<
+      string,
+      unknown
+    >;
+    expect(replyContexts.has('msg-old')).toBe(false);
+    expect(msgSeqMap.has('msg-old')).toBe(false);
+    expect(replyContexts.has('msg-latest')).toBe(true);
+    expect(msgSeqMap.get('msg-latest')).toBe(1);
+    ch.disconnect();
+  });
+
   it('does not call saveQQState when no entries are evicted', () => {
     vi.useFakeTimers();
     const ch = makeChannel();

packages/channels/qqbot/src/stream.test.ts

@@ -52,16 +52,23 @@ vi.mock('@qwen-code/channel-base', () => ({
     protected handleInbound(_env: unknown): Promise<void> {
       return Promise.resolve();
     }
+    protected getResponseMessageId(_sessionId: string): string | undefined {
+      return undefined;
+    }
     protected async onResponseComplete(
       _chatId: string,
       fullText: string,
-      _sessionId: string,
+      sessionId: string,
     ): Promise<void> {
       await (
         this as unknown as {
-          sendMessage: (c: string, t: string) => Promise<void>;
+          sendResponseMessage: (
+            c: string,
+            t: string,
+            s: string,
+          ) => Promise<void>;
         }
-      ).sendMessage(_chatId, fullText);
+      ).sendResponseMessage(_chatId, fullText, sessionId);
     }
     onSessionDied(_sessionId: string): void {
       // no-op in mock; overridden by QQChannel
@@ -135,12 +142,23 @@ function onResponseChunk(
   chatId: string,
   chunk: string,
   sessionId: string,
+  messageId?: string,
 ) {
   return (
     ch as unknown as {
-      onResponseChunk: (c: string, h: string, s: string) => void;
+      onResponseChunk: (
+        c: string,
+        h: string,
+        s: string,
+        segment?: { messageId?: string },
+      ) => void;
     }
-  ).onResponseChunk(chatId, chunk, sessionId);
+  ).onResponseChunk(
+    chatId,
+    chunk,
+    sessionId,
+    messageId ? { messageId } : undefined,
+  );
 }
 
 function onResponseComplete(
@@ -333,6 +351,52 @@ describe('idle-flush timer', () => {
     await drain();
     expect(mockSendQQMessage).toHaveBeenCalledTimes(2);
   });
+
+  it('keeps same-chat streams bound to their originating messages', async () => {
+    const ch = makeChannel();
+    const replyContexts = (
+      ch as unknown as {
+        replyContextByMessageId: Map<
+          string,
+          { chatId: string; msgId: string; timestamp: number }
+        >;
+      }
+    ).replyContextByMessageId;
+    const timestamp = Date.now();
+    replyContexts.set('msg-a', {
+      chatId: 'test-chat',
+      msgId: 'msg-a',
+      timestamp,
+    });
+    replyContexts.set('msg-b', {
+      chatId: 'test-chat',
+      msgId: 'msg-b',
+      timestamp,
+    });
+
+    onResponseChunk(ch, 'test-chat', 'a-1', 'session-a', 'msg-a');
+    onResponseChunk(ch, 'test-chat', 'b-1', 'session-b', 'msg-b');
+    vi.advanceTimersByTime(2000);
+    await drain();
+
+    expect(mockSendQQMessage).toHaveBeenCalledTimes(2);
+    expect(mockSendQQMessage.mock.calls[0][3]).toMatchObject({
+      msg_id: 'msg-a',
+      msg_seq: 1,
+    });
+    expect(mockSendQQMessage.mock.calls[1][3]).toMatchObject({
+      msg_id: 'msg-b',
+      msg_seq: 1,
+    });
+
+    onResponseChunk(ch, 'test-chat', 'a-2', 'session-a', 'msg-a');
+    vi.advanceTimersByTime(2000);
+    await drain();
+    expect(mockSendQQMessage.mock.calls[2][3]).toMatchObject({
+      msg_id: 'msg-a',
+      msg_seq: 2,
+    });
+  });
 });
 
 describe('onToolCall flush', () => {

packages/channels/weixin/src/WeixinAdapter.test.ts

@@ -109,12 +109,14 @@ describe('WeixinChannel', () => {
     expect(setTyping).toHaveBeenCalledTimes(2);
   });
 
-  it('clears failed start typing state so a later started event can retry', async () => {
+  it('retries a failed initial typing request while the session is active', async () => {
     const channel = createChannel();
     const chatId = 'user-retry';
-    const activeTypingChats = (
-      channel as unknown as { activeTypingChats: Set<string> }
-    ).activeTypingChats;
+    const activeTypingSessions = (
+      channel as unknown as {
+        activeTypingSessions: Map<string, Map<string, number>>;
+      }
+    ).activeTypingSessions;
 
     apiMocks.getConfig.mockResolvedValue({ typing_ticket: 'ticket-1' });
     apiMocks.sendTyping
@@ -134,17 +136,48 @@ describe('WeixinChannel', () => {
 
     await vi.waitFor(() => {
       expect(apiMocks.sendTyping).toHaveBeenCalledTimes(1);
-      expect(activeTypingChats.has(chatId)).toBe(false);
+      expect(activeTypingSessions.get(chatId)?.has('session-2')).toBe(true);
     });
 
-    channel.emitLifecycle({ ...baseEvent, type: 'started' });
-
+    await vi.advanceTimersByTimeAsync(4000);
     await vi.waitFor(() => {
       expect(apiMocks.sendTyping).toHaveBeenCalledTimes(2);
-      expect(activeTypingChats.has(chatId)).toBe(true);
     });
   });
 
+  it('keeps typing until every session in the same chat terminates', () => {
+    const channel = createChannel();
+    const setTyping = vi.fn().mockResolvedValue(true);
+    (channel as unknown as { setTyping: typeof setTyping }).setTyping =
+      setTyping;
+    const base = {
+      channelName: 'weixin',
+      chatId: 'shared-chat',
+      messageId: 'message',
+      identity: { id: 'channel:weixin', displayName: 'weixin' },
+      memoryScope: { namespace: 'channel:weixin', mode: 'metadata-only' },
+    } satisfies Omit<LifecycleBase, 'sessionId'>;
+
+    channel.emitLifecycle({ ...base, sessionId: 'session-a', type: 'started' });
+    channel.emitLifecycle({ ...base, sessionId: 'session-b', type: 'started' });
+    channel.emitLifecycle({
+      ...base,
+      sessionId: 'session-a',
+      type: 'completed',
+    });
+
+    expect(setTyping).toHaveBeenCalledTimes(1);
+    expect(setTyping).toHaveBeenLastCalledWith('shared-chat', true);
+
+    channel.emitLifecycle({
+      ...base,
+      sessionId: 'session-b',
+      type: 'completed',
+    });
+    expect(setTyping).toHaveBeenCalledTimes(2);
+    expect(setTyping).toHaveBeenLastCalledWith('shared-chat', false);
+  });
+
   it('stops typing again when a late lifecycle start resolves after terminal cleanup', async () => {
     const channel = createChannel();
     const start = deferredPromise<boolean>();
@@ -183,9 +216,11 @@ describe('WeixinChannel', () => {
     const setTyping = vi.fn().mockResolvedValue(true);
     (channel as unknown as { setTyping: typeof setTyping }).setTyping =
       setTyping;
-    const activeTypingChats = (
-      channel as unknown as { activeTypingChats: Set<string> }
-    ).activeTypingChats;
+    const activeTypingSessions = (
+      channel as unknown as {
+        activeTypingSessions: Map<string, Map<string, number>>;
+      }
+    ).activeTypingSessions;
 
     channel.emitLifecycle({
       type: 'started',
@@ -196,11 +231,11 @@ describe('WeixinChannel', () => {
       identity: { id: 'channel:weixin', displayName: 'weixin' },
       memoryScope: { namespace: 'channel:weixin', mode: 'metadata-only' },
     });
-    expect(activeTypingChats.has('user-disconnect')).toBe(true);
+    expect(activeTypingSessions.has('user-disconnect')).toBe(true);
 
     channel.disconnect();
 
-    expect(activeTypingChats.has('user-disconnect')).toBe(false);
+    expect(activeTypingSessions.has('user-disconnect')).toBe(false);
   });
 
   describe('typing keepalive', () => {
@@ -246,7 +281,7 @@ describe('WeixinChannel', () => {
       expect(setTyping).toHaveBeenCalledTimes(5);
     });
 
-    it('does not arm the keepalive when the initial TYPING fails', async () => {
+    it('retries through keepalive when the initial TYPING fails', async () => {
       const channel = createChannel();
       const setTyping = vi.fn().mockResolvedValue(false);
       installSetTypingMock(channel, setTyping);
@@ -255,7 +290,7 @@ describe('WeixinChannel', () => {
       channel.emitLifecycle({ ...base, type: 'started' });
       await vi.advanceTimersByTimeAsync(20000);
 
-      expect(setTyping).toHaveBeenCalledTimes(1);
+      expect(setTyping).toHaveBeenCalledTimes(6);
       expect(setTyping).toHaveBeenLastCalledWith('user-keepalive-fail', true);
     });
 
@@ -464,6 +499,53 @@ describe('WeixinChannel', () => {
       expect(setTyping).toHaveBeenCalledTimes(callsAtBackstop);
     });
 
+    it('expires an old session without cancelling a newer session in the same chat', async () => {
+      const channel = createChannel();
+      const setTyping = vi.fn().mockResolvedValue(true);
+      installSetTypingMock(channel, setTyping);
+      const oldSession = keepaliveEvent('user-shared', 'session-old');
+      const newSession = keepaliveEvent('user-shared', 'session-new');
+
+      channel.emitLifecycle({ ...oldSession, type: 'started' });
+      await vi.advanceTimersByTimeAsync(TYPING_KEEPALIVE_MAX_MS - 4000);
+      channel.emitLifecycle({ ...newSession, type: 'started' });
+
+      await vi.advanceTimersByTimeAsync(4000);
+      expect(setTyping).toHaveBeenLastCalledWith('user-shared', true);
+
+      const activeTypingSessions = (
+        channel as unknown as {
+          activeTypingSessions: Map<string, Map<string, number>>;
+        }
+      ).activeTypingSessions;
+      expect(activeTypingSessions.get('user-shared')?.has('session-old')).toBe(
+        false,
+      );
+      expect(activeTypingSessions.get('user-shared')?.has('session-new')).toBe(
+        true,
+      );
+
+      channel.emitLifecycle({ ...newSession, type: 'completed' });
+      expect(setTyping).toHaveBeenLastCalledWith('user-shared', false);
+    });
+
+    it('removes only the dead session from a shared chat', () => {
+      const channel = createChannel();
+      const setTyping = vi.fn().mockResolvedValue(true);
+      installSetTypingMock(channel, setTyping);
+      const sessionA = keepaliveEvent('user-session-died', 'session-a');
+      const sessionB = keepaliveEvent('user-session-died', 'session-b');
+
+      channel.emitLifecycle({ ...sessionA, type: 'started' });
+      channel.emitLifecycle({ ...sessionB, type: 'started' });
+      channel.onSessionDied('session-a');
+      expect(setTyping).toHaveBeenCalledTimes(1);
+
+      channel.onSessionDied('session-b');
+      expect(setTyping).toHaveBeenCalledTimes(2);
+      expect(setTyping).toHaveBeenLastCalledWith('user-session-died', false);
+    });
+
     it('ignores a stale failed initial TYPING from a previous turn', async () => {
       const channel = createChannel();
       const stale = deferredPromise<boolean>();
@@ -594,13 +676,13 @@ describe('WeixinChannel', () => {
       const priv = channel as unknown as {
         typingKeepaliveIntervals: Map<string, unknown>;
         typingKeepaliveInFlight: Set<string>;
-        typingKeepaliveArmedAt: Map<string, number>;
         typingGenerations: Map<string, number>;
+        activeTypingSessions: Map<string, Map<string, number>>;
       };
       expect(priv.typingKeepaliveIntervals.size).toBe(0);
       expect(priv.typingKeepaliveInFlight.size).toBe(0);
-      expect(priv.typingKeepaliveArmedAt.size).toBe(0);
       expect(priv.typingGenerations.size).toBe(0);
+      expect(priv.activeTypingSessions.size).toBe(0);
 
       // A fresh turn on the reused adapter must not be reaped early by
       // stale backstop state: it should refresh, not CANCEL.

packages/channels/weixin/src/WeixinAdapter.ts

@@ -52,13 +52,12 @@ function escapeRegex(s: string): string {
 
 export class WeixinChannel extends ChannelBase {
   private abortController: AbortController | null = null;
-  private activeTypingChats = new Set<string>();
+  private activeTypingSessions = new Map<string, Map<string, number>>();
   private typingKeepaliveIntervals = new Map<
     string,
     ReturnType<typeof setInterval>
   >();
   private typingKeepaliveInFlight = new Set<string>();
-  private typingKeepaliveArmedAt = new Map<string, number>();
   /**
    * Per-chat generation token; stale setTyping continuations bail out.
    * Entries are reclaimed only by `disconnect()` — deleting them in
@@ -162,21 +161,21 @@ export class WeixinChannel extends ChannelBase {
     );
   }
 
-  protected override onPromptStart(chatId: string): void {
-    this.startTyping(chatId);
+  protected override onPromptStart(chatId: string, sessionId: string): void {
+    this.startTyping(chatId, sessionId);
   }
 
-  protected override onPromptEnd(chatId: string): void {
-    this.stopTyping(chatId);
+  protected override onPromptEnd(chatId: string, sessionId: string): void {
+    this.stopTyping(chatId, sessionId);
   }
 
   protected override onTaskLifecycle(event: ChannelTaskLifecycleEvent): void {
     if (event.type === 'started') {
-      this.startTyping(event.chatId);
+      this.startTyping(event.chatId, event.sessionId);
       return;
     }
     if (isTerminalTaskLifecycleType(event.type)) {
-      this.stopTyping(event.chatId);
+      this.stopTyping(event.chatId, event.sessionId);
     }
   }
 
@@ -326,14 +325,19 @@ export class WeixinChannel extends ChannelBase {
     }
     this.typingKeepaliveIntervals.clear();
     this.typingKeepaliveInFlight.clear();
-    this.typingKeepaliveArmedAt.clear();
     this.typingGenerations.clear();
-    this.activeTypingChats.clear();
+    this.activeTypingSessions.clear();
   }
 
-  private startTyping(chatId: string): void {
-    if (this.activeTypingChats.has(chatId)) return;
-    this.activeTypingChats.add(chatId);
+  private startTyping(chatId: string, sessionId: string): void {
+    const sessions =
+      this.activeTypingSessions.get(chatId) ?? new Map<string, number>();
+    if (sessions.has(sessionId)) return;
+    const firstSession = sessions.size === 0;
+    sessions.set(sessionId, Date.now());
+    this.activeTypingSessions.set(chatId, sessions);
+    if (!firstSession) return;
+
     const controller = this.abortController;
     const generation = (this.typingGenerations.get(chatId) ?? 0) + 1;
     this.typingGenerations.set(chatId, generation);
@@ -347,69 +351,75 @@ export class WeixinChannel extends ChannelBase {
       // must not touch the live turn's typing state. A late success still
       // re-set the server-side indicator after our CANCEL, so compensate —
       // but only while the chat is idle; if a successor turn is live
-      // (activeTypingChats re-added), an unconditional CANCEL here would
+      // (activeTypingSessions re-added), an unconditional CANCEL here would
       // blank the successor's indicator mid-turn. A late failure is
       // harmless and must not delete live state.
       if (generation !== this.typingGenerations.get(chatId)) {
-        if (started && !this.activeTypingChats.has(chatId)) {
+        if (started && !this.activeTypingSessions.has(chatId)) {
           void this.setTyping(chatId, false);
         }
         return;
       }
-      if (!started) {
-        this.activeTypingChats.delete(chatId);
-        return;
-      }
-      if (!this.activeTypingChats.has(chatId)) {
+      if (started && !this.activeTypingSessions.has(chatId)) {
         void this.setTyping(chatId, false);
         return;
       }
-      this.startTypingKeepalive(chatId);
+      if (this.activeTypingSessions.has(chatId)) {
+        this.startTypingKeepalive(chatId);
+      }
     });
   }
 
-  private stopTyping(chatId: string): void {
+  private stopTyping(chatId: string, sessionId: string): void {
+    const sessions = this.activeTypingSessions.get(chatId);
+    if (!sessions?.delete(sessionId)) return;
+    if (sessions.size > 0) return;
+    this.activeTypingSessions.delete(chatId);
+    this.stopTypingChat(chatId);
+  }
+
+  private stopTypingChat(chatId: string): void {
     // Invalidate any in-flight initial TYPING of a previous turn.
     this.typingGenerations.set(
       chatId,
       (this.typingGenerations.get(chatId) ?? 0) + 1,
     );
     this.stopTypingKeepalive(chatId);
-    if (!this.activeTypingChats.delete(chatId)) return;
     void this.setTyping(chatId, false);
   }
 
   /**
    * Refreshes TYPING periodically while the chat stays in
-   * `activeTypingChats`, so the indicator survives long turns. The interval
-   * is only armed after the initial TYPING is confirmed, and each tick
-   * self-reaps once the chat is no longer typing or the backstop
-   * (`TYPING_KEEPALIVE_MAX_MS`) elapses, so a missed terminal event cannot
-   * keep it alive indefinitely.
+   * active sessions, so the indicator survives long turns. Each tick expires
+   * only sessions that exceeded the backstop; newer sessions in the same chat
+   * keep ownership.
    */
   private startTypingKeepalive(chatId: string): void {
     if (this.typingKeepaliveIntervals.has(chatId)) return;
-    this.typingKeepaliveArmedAt.set(chatId, Date.now());
     this.typingKeepaliveIntervals.set(
       chatId,
       setInterval(() => {
-        if (!this.activeTypingChats.has(chatId)) {
+        const sessions = this.activeTypingSessions.get(chatId);
+        if (!sessions) {
           this.stopTypingKeepalive(chatId);
           return;
         }
-        const armedAt = this.typingKeepaliveArmedAt.get(chatId);
-        if (
-          armedAt !== undefined &&
-          Date.now() - armedAt >= TYPING_KEEPALIVE_MAX_MS
-        ) {
-          // Wedged turn — bound the leak and clear the indicator. The
-          // backstop firing is the only observable signal that a turn never
-          // emitted a terminal event, so leave a log line tying the dropped
-          // indicator to the wedged chat.
+        const now = Date.now();
+        let expiredSessions = 0;
+        for (const [sessionId, startedAt] of sessions) {
+          if (now - startedAt >= TYPING_KEEPALIVE_MAX_MS) {
+            sessions.delete(sessionId);
+            expiredSessions++;
+          }
+        }
+        if (expiredSessions > 0) {
           process.stderr.write(
-            `[Weixin:${this.name}] Typing keepalive backstop (${TYPING_KEEPALIVE_MAX_MS}ms) elapsed for chat ${chatId} without a terminal event; clearing the indicator\n`,
+            `[Weixin:${this.name}] Typing keepalive backstop (${TYPING_KEEPALIVE_MAX_MS}ms) elapsed for ${expiredSessions} session(s) in chat ${chatId}\n`,
           );
-          this.stopTyping(chatId);
+        }
+        if (sessions.size === 0) {
+          this.activeTypingSessions.delete(chatId);
+          this.stopTypingChat(chatId);
           return;
         }
         // Don't overlap keepalive requests for the same chat.
@@ -427,7 +437,6 @@ export class WeixinChannel extends ChannelBase {
     if (interval === undefined) return;
     clearInterval(interval);
     this.typingKeepaliveIntervals.delete(chatId);
-    this.typingKeepaliveArmedAt.delete(chatId);
     // A refresh in flight from the ending turn would otherwise keep the flag
     // set until it settles (up to the post() timeout), and every keepalive
     // tick of the successor turn would early-return on the dedup check —
@@ -436,6 +445,15 @@ export class WeixinChannel extends ChannelBase {
     this.typingKeepaliveInFlight.delete(chatId);
   }
 
+  override onSessionDied(sessionId: string): void {
+    for (const [chatId, sessions] of this.activeTypingSessions) {
+      if (sessions.has(sessionId)) {
+        this.stopTyping(chatId, sessionId);
+      }
+    }
+    super.onSessionDied(sessionId);
+  }
+
   private async setTyping(userId: string, typing: boolean): Promise<boolean> {
     try {
       let ticket = typingTickets.get(userId);