添加 /stop 抢占功能以立即中止当前消息、清空排队消息

[JingWwwwwj]

钉钉插件/stop 抢跑 —— 代码改动说明

本文档说明为支持 /stop 即时打断当前任务而对钉钉插件做的所有代码改动：

1. 改了什么
2. 为什么这么改
3. 实现了什么效果，以及修改前后的结果对比。
4. 以上修改基于钉钉官方出品的 OpenClaw 钉钉 Channel 插件 v0.8.22：https://github.com/DingTalk-Real-AI/dingtalk-openclaw-connector

---

一、问题回顾

用户通过钉钉给 OpenClaw 发了一条耗时很长的消息 msg1，agent 正在生成；后续 msg2…msgN 已经排在本地串行队列里。此时用户想终止，发了 /stop。

期望的行为：/stop 立即抢占当前 run，触发 abort，并清空所有待处理消息。

修改前的实际行为：/stop 被当成普通消息塞进同一条 promise 串行链尾部，必须等 msg1…msgN 全部处理完才轮到自己执行；而那时 msg1 早已自然结束，abort 时已无事可做 —— stop 等于没按。

---

二、改动总览

只修改了 dingtalk 插件源码中的一个文件：

dingtalk-openclaw-connector-main/src/core/message-handler.ts

序号	位置	改动	作用
1	第 107–116 行	新增 isStopCommand() 函数	识别 /stop 或 stop
2	第 98–105 行	新增 sessionGenerations 代际数	辅助"清空待处理队列"
3	第 1724–1752 行（handleDingTalkMessage 入口）	加 stop 抢跑预检	stop 不入队，直接处理
4	第 1815–1826 行（入队任务体内）	加代际数的判断，跳过待处理队列	已排队但被 stop 作废的消息自动跳过
5	第 118–127 行（TTL 清理）	同步清理代际表	-

---

三、详细改动

改动 1：新增 isStopCommand() 函数

位置：message-handler.ts:107-116

/**
 * 严格匹配 "/stop" 或 "stop"。
 * 必须是 OpenClaw isAbortRequestText 的子集 —— 命中即抢跑，未命中走原队列。
 * 不能放宽，否则会出现"绕了队列但 SDK 不 abort"的事故。
 */
function isStopCommand(text: string): boolean {
  if (!text) return false;
  const normalized = text.trim().toLowerCase().replace(/\s+/g, " ");
  return normalized === "/stop" || normalized === "stop";
}

为什么这么写：

• 所有消息到达 openclaw 后，openclaw 都会再判断一下是不是它所规定的暂停命令
• 所以抢跑路径必须保证：本地判 "是 stop" 时，OpenClaw SDK 也判 "是 stop"。否则会出现"绕过队列、却没触发 abort、最终变成普通消息插队执行"。
• 只匹配 /stop 和 stop，这两个是 openclaw 官方 SDK isAbortRequestText 的真子集，绝对安全。
• 像"停一下""暂停"这种词，会按照原排队流程，体验不会变差。

效果：拿到一条文本立即就能判断"要不要抢跑"，没什么副作用。

---

改动 2：新增 sessionGenerations 代际计数表

位置：message-handler.ts:98-105

/**
 * /stop 抢跑代际计数。
 * 每个 queueKey 独立递增；每条入队任务在 .then 体内核对自己捕获的代际，
 * 不一致就跳过 —— 用于在 /stop 抢跑时丢弃所有已排队但尚未执行的 msg2..msgN。
 * (sessionQueues 只是 Promise 链的引用 Map，仅 delete 不会取消已经构造的 .then 回调，
 * 因此必须配合代际检查才能真正"清空待处理消息队列"。)
 */
const sessionGenerations = new Map<string, number>();

为什么这么写：

JavaScript 的 promise 链有一个关键性质：Map 里的 promise 引用只是"队列尾巴的句柄"。.delete() 只是丢掉句柄，并不会取消已经挂在 promise 链上的 **.then** 回调。msg2…msgN 还会被原链按顺序执行完。

要真正"清空待处理队列"，必须在每个 .then 回调里加一个"我还该不该执行"的检查：

• 每条消息入队时记下 taskGeneration = sessionGenerations.get(queueKey)
• 真正执行前再读一次 currentGen，不一致就跳过该消息直接不处理
• /stop 来时把 sessionGenerations[queueKey] +1，所有"过期代际"的待执行任务自动被废掉

效果：把"清队列"变成了"每条消息出队前自检"，但实现效果上完全等价于队列被清空。

---

改动 3：入口加 /stop 检查

位置：message-handler.ts:1724-1752

算完 queueKey 之后、原入队代码之前：

// ===== /stop 抢跑：识别到 stop 命令时不入队，直接走 SDK abort 链路 =====
// 设计要点：
//  1) 提升代际 sessionGenerations[queueKey] += 1，让所有已排队但未执行的 task
//     在自己的 .then 体内检测到代际不匹配并 return，等价于"清空待处理消息队列"。
//  2) 不调 sessionQueues.delete(queueKey)：in-flight 的 msg1 仍持有 currentTask，
//     让它在 .finally 自然清理；同时 stop 之后到达的 msg 会自然挂在原链尾部，
//     不破坏同 session 的串行化。
//  3) 抢跑路径直接 await handleDingTalkMessageInternal(...)，与 in-flight 的 msg1
//     在 event loop 中并发执行；内部走到 dispatchReplyFromConfig 时，SDK 第一步
//     的 tryFastAbortFromMessage 会调用 abortSessionRunTarget，触发 msg1 的
//     AbortSignal，并自动回 "⚙️ Agent was aborted."。
const inboundText = extractMessageContent(data).text?.trim() ?? "";
if (inboundText && isStopCommand(inboundText)) {
  const nextGen = (sessionGenerations.get(queueKey) ?? 0) + 1;
  sessionGenerations.set(queueKey, nextGen);
  sessionLastActivity.set(queueKey, Date.now());
  log?.info?.(`[stop 抢跑] queueKey=${queueKey} gen->${nextGen} text=${inboundText.slice(0, 50)}`);

  try {
    await handleDingTalkMessageInternal({
      ...params,
      preCreatedCard: undefined,
      emotionAlreadyAdded: false,
    });
  } catch (err: any) {
    log?.error?.(`[stop 抢跑] 异常：${err?.message ?? err}`);
  }
  return;
}
// ===== 以下代码是原有入队流程 =====

设计决策说明：

决策	原因
不入队，直接 await handleDingTalkMessageInternal(...)	进 event loop 与 msg1 并发，才有机会在 msg1 还活着的时候 abort 它
提升代际 而不是 sessionQueues.delete()	见改动 2：删 Map 不会取消已挂的 .then，必须靠代际检查一下
preCreatedCard: undefined + emotionAlreadyAdded: false	stop 是新分流路径，没有上游 ACK 卡片或表情，必须显式传干净的初始状态
try / catch / return	抢跑失败也只影响这一条 stop 自己；不能让异常冒泡污染原队列

效果：

钉钉 webhook → handleDingTalkMessage
      → isStopCommand 命中 → 不入队
      → handleDingTalkMessageInternal
      → core.channel.reply.dispatchReplyFromConfig（以下为 Openclaw 的 SDK 处理）
      → tryFastAbortFromMessage
      → abortSessionRunTarget(msg1 的 sessionId)
      → msg1 的 AbortSignal 触发，run 中断
      → SDK 自动回 "⚙️ Agent was aborted."

---

改动 4：入队任务之前加代际检查

位置：message-handler.ts:1815-1826

// 入队时同步捕获当前代际；轮到自己执行时核对，
// 若已被 /stop 抢跑提升过，则跳过本条消息（实现"清空待处理队列"语义）。
const taskGeneration = sessionGenerations.get(queueKey) ?? 0;

// 创建当前消息的处理任务
const currentTask = previousTask
  .then(async () => {
    const currentGen = sessionGenerations.get(queueKey) ?? 0;
    if (currentGen !== taskGeneration) {
      log?.info?.(`[队列] 消息已被 /stop 取消，跳过 queueKey=${queueKey} gen=${taskGeneration}->${currentGen}`);
      return;
    }
    log?.info?.(`[队列] 开始处理消息，queueKey=${queueKey}`);
    await handleDingTalkMessageInternal({ ...params, preCreatedCard, emotionAlreadyAdded: isQueueBusy });
    log?.info?.(`[队列] 消息处理完成，queueKey=${queueKey}`);
  })
  // .catch / .finally 保持原样

为什么这么写：

• taskGeneration 在入队那一刻就确定
• 真正轮到执行前先看现在是哪一代 —— 只要中间发生过 /stop（代际 +1），就跳过
• 跳过的消息只是不调handleDingTalkMessageInternal，promise 链本身仍然推进，后续消息照常出队

效果：和改动 3 的"代际数增加"配合，才让 stop 具有"清空已排队消息"的能力。

---

改动 5：TTL 清理同步带上代际表

位置：message-handler.ts:118-127

function cleanupExpiredSessionQueues(): void {
  const now = Date.now();
  for (const [queueKey, lastActivity] of sessionLastActivity.entries()) {
    if (now - lastActivity > SESSION_QUEUE_TTL) {
      sessionQueues.delete(queueKey);
      sessionLastActivity.delete(queueKey);
      sessionGenerations.delete(queueKey);   // ← 新增这一行
    }
  }
}

为什么需要：新增了 sessionGenerations 这个 Map，需要挂到现有 5 分钟 TTL 清理逻辑里。

---

四、修改前后行为对比

任务场景：同一时刻有 msg1 在跑、msg2-msg5 在排队，用户发 /stop

修改前

event loop 时间线（横轴）：

msg1 ─────────────────────────────────────► 自然结束
                                                    │
                                                    ▼
                                                  msg2 ─► msg3 ─► msg4 ─► msg5
                                                                                  │
                                                                                  ▼
                                                                                /stop ─► (调 abort，但已无事可干)

• /stop 排在最后，必须等前面 5 条全跑完
• 等到它，msg1 早已结束，abort 无意义
• 用户体感：发了 stop 完全没反应，agent 仍在输出

修改后

event loop 时间线（横轴）：

msg1 ──────────abort──► 中断，回复 "⚙️ Agent was aborted."
                  ▲
                  │  并发触发
                  │
/stop（不入队）──┘

msg2..msg5（已排队但代际过期）──► 出队时自检失败，直接 return，不调 internal

• /stop 抢跑，与 msg1 在 event loop 里并发
• tryFastAbortFromMessage 触发 msg1 的 AbortSignal，run 立即中断
• msg2..msg5 在轮到自己时检测到代际过期，自动跳过
• 用户体感：暂停当前任务，收到 "⚙️ Agent was aborted."

普通消息路径不变（仍走串行队列），仅多了一道"代际是否过期"的轻量检查。

---

五、未改动的部分

• OpenClaw 的代码
• 普通消息的入队/出队/AI Card ACK 逻辑 —— 仅在出队时多了一行代际自检，行为不变

[meng93]

谢谢 PR，设计思路讲得很清楚。Review 下来有几个问题需要处理：

P0 必修

isStopCommand 代码与文档不一致

JSDoc 和 PR 说明都强调：「必须是 OpenClaw isAbortRequestText 的子集」，文档里只列了 /stop / stop，但代码里多了 || normalized === "停止"。如果 OpenClaw SDK 里「停止」不是 abort 词，正好命中 JSDoc 里警告的「绕了队列但 SDK 不 abort」事故。

请删掉 "停止" 分支，保持与文档 / SDK 子集承诺一致。

P1 必补

本 PR 没有任何测试

这块涉及 Promise 链 + 代际计数 + 抢跑并发，逻辑细微，必须有单测覆盖以下场景后再合：

• isStopCommand 正负样本（含 /Stop、大小写、多空格、/stop please 这种不应命中的）
• /stop 抢跑时：in-flight msg1 被 abort，msg2..msgN 跳过
• /stop 在队列空闲时到达的行为
• /stop 之后到达的新消息代际正确、不被旧 stop 误杀
• TTL 清理后 sessionGenerations 同步清空

P2 建议另开 issue 跟进（不阻塞本 PR）

1. 抢跑分支里 extractMessageContent(data) 被调了两次（抢跑判断一次 + 进 internal 后再一次），可以把第一次结果通过 params 透下去。

2. 抢跑分支的 catch 静默吞异常，用户端拿不到任何反馈。体验上万一 abort 本身挂了等同于「stop 没按」，是否需要兜底文案可以另讨。

3. 日志 tag [stop抢跑] 与项目里其他位置可能的 [DingTalk][...] 结构化前缀不一致。

4. 新增代码位于原 try 块之前，extractMessageContent 如果对畸形 data 会抛，会引入未捕获异常路径。需核实该函数在意外输入下是否会 throw。

P0 + P1 处理完后可以再丢上来。一天内合入