refactor(reply): type reply threading policy

steipete · steipete · commit 9b7002ee5971 · 2026-04-05T21:40:56.000+01:00
diff --git a/docs/.generated/plugin-sdk-api-baseline.sha256 b/docs/.generated/plugin-sdk-api-baseline.sha256
@@ -1,2 +1,2 @@
-1a70d4d4f34ba5d0708a17540c0cbf1c98f50d37f25d2f71ad99b8bf6856cf9b  plugin-sdk-api-baseline.json
-99cbe863efbed5ab42e0e7053d9486179aa689807696f0ebc4f4b89f1fe8cdfd  plugin-sdk-api-baseline.jsonl
+97509287d728c8f5d1736f7ea07521451ada4b9d7ef56555dbe860a89e1b6e08  plugin-sdk-api-baseline.json
+a22b3d427953cc8394b28c87ef7a992d2eb4f2c9f6a76fa58b33079e2306661b  plugin-sdk-api-baseline.jsonl
diff --git a/docs/channels/slack.md b/docs/channels/slack.md
@@ -338,7 +338,7 @@ Current Slack message actions include `send`, `upload-file`, `download-file`, `r
 
 Reply threading controls:
 
-- `channels.slack.replyToMode`: `off|first|all` (default `off`)
+- `channels.slack.replyToMode`: `off|first|all|batched` (default `off`)
 - `channels.slack.replyToModeByChatType`: per `direct|group|channel`
 - legacy fallback for direct chats: `channels.slack.dm.replyToMode`
 
diff --git a/docs/gateway/configuration-reference.md b/docs/gateway/configuration-reference.md
@@ -179,7 +179,7 @@ WhatsApp runs through the gateway's web channel (Baileys Web). It starts automat
         { command: "generate", description: "Create an image" },
       ],
       historyLimit: 50,
-      replyToMode: "first", // off | first | all
+      replyToMode: "first", // off | first | all | batched
       linkPreview: true,
       streaming: "partial", // off | partial | block | progress (default: off; opt in explicitly to avoid preview-edit rate limits)
       actions: { reactions: true, sendMessage: true },
@@ -239,7 +239,7 @@ WhatsApp runs through the gateway's web channel (Baileys Web). It starts automat
         events: true,
         moderation: false,
       },
-      replyToMode: "off", // off | first | all
+      replyToMode: "off", // off | first | all | batched
       dmPolicy: "pairing",
       allowFrom: ["1234567890", "123456789012345678"],
       dm: { enabled: true, groupEnabled: false, groupChannels: ["openclaw-dm"] },
@@ -405,7 +405,7 @@ WhatsApp runs through the gateway's web channel (Baileys Web). It starts automat
       allowBots: false,
       reactionNotifications: "own",
       reactionAllowlist: ["U123"],
-      replyToMode: "off", // off | first | all
+      replyToMode: "off", // off | first | all | batched
       thread: {
         historyScope: "thread", // thread | channel
         inheritParent: false,
diff --git a/extensions/discord/src/monitor/message-handler.batch-gate.test.ts b/extensions/discord/src/monitor/message-handler.batch-gate.test.ts
@@ -5,18 +5,18 @@ describe("applyImplicitReplyBatchGate", () => {
   it("leaves context unchanged when replyToMode is not batched", () => {
     const ctx: Record<string, unknown> = {};
     applyImplicitReplyBatchGate(ctx, "first", true);
-    expect(ctx.AllowImplicitReplyToCurrentMessage).toBeUndefined();
+    expect(ctx.ReplyThreading).toBeUndefined();
   });
 
   it("marks single-message turns as not eligible for implicit reply refs", () => {
     const ctx: Record<string, unknown> = {};
     applyImplicitReplyBatchGate(ctx, "batched", false);
-    expect(ctx.AllowImplicitReplyToCurrentMessage).toBe(false);
+    expect(ctx.ReplyThreading).toEqual({ implicitCurrentMessage: "deny" });
   });
 
   it("marks batched turns as eligible for implicit reply refs", () => {
     const ctx: Record<string, unknown> = {};
     applyImplicitReplyBatchGate(ctx, "batched", true);
-    expect(ctx.AllowImplicitReplyToCurrentMessage).toBe(true);
+    expect(ctx.ReplyThreading).toEqual({ implicitCurrentMessage: "allow" });
   });
 });
diff --git a/extensions/discord/src/monitor/message-handler.batch-gate.ts b/extensions/discord/src/monitor/message-handler.batch-gate.ts
@@ -1,12 +1,19 @@
 import type { ReplyToMode } from "openclaw/plugin-sdk/config-runtime";
+import type { ReplyThreadingPolicy } from "openclaw/plugin-sdk/reply-reference";
+import { resolveBatchedReplyThreadingPolicy } from "openclaw/plugin-sdk/reply-reference";
 
-export function applyImplicitReplyBatchGate(
-  ctx: Record<string, unknown>,
+type ReplyThreadingContext = {
+  ReplyThreading?: ReplyThreadingPolicy;
+};
+
+export function applyImplicitReplyBatchGate<T extends object>(
+  ctx: T,
   replyToMode: ReplyToMode,
   isBatched: boolean,
 ) {
-  if (replyToMode !== "batched") {
+  const replyThreading = resolveBatchedReplyThreadingPolicy(replyToMode, isBatched);
+  if (!replyThreading) {
     return;
   }
-  ctx.AllowImplicitReplyToCurrentMessage = isBatched;
+  (ctx as T & ReplyThreadingContext).ReplyThreading = replyThreading;
 }
diff --git a/extensions/discord/src/monitor/message-handler.ts b/extensions/discord/src/monitor/message-handler.ts
@@ -145,11 +145,7 @@ export function createDiscordMessageHandler(
         if (!ctx) {
           return;
         }
-        applyImplicitReplyBatchGate(
-          ctx as unknown as Record<string, unknown>,
-          params.replyToMode,
-          false,
-        );
+        applyImplicitReplyBatchGate(ctx, params.replyToMode, false);
         inboundWorker.enqueue(buildDiscordInboundJob(ctx));
         return;
       }
@@ -182,11 +178,7 @@ export function createDiscordMessageHandler(
       if (!ctx) {
         return;
       }
-      applyImplicitReplyBatchGate(
-        ctx as unknown as Record<string, unknown>,
-        params.replyToMode,
-        true,
-      );
+      applyImplicitReplyBatchGate(ctx, params.replyToMode, true);
       if (entries.length > 1) {
         const ids = entries.map((entry) => entry.data.message?.id).filter(Boolean) as string[];
         if (ids.length > 0) {
diff --git a/extensions/discord/src/monitor/reply-delivery.ts b/extensions/discord/src/monitor/reply-delivery.ts
@@ -9,6 +9,7 @@ import {
   resolveTextChunksWithFallback,
   sendMediaWithLeadingCaption,
 } from "openclaw/plugin-sdk/reply-payload";
+import { isSingleUseReplyToMode } from "openclaw/plugin-sdk/reply-reference";
 import {
   resolveRetryConfig,
   retryAsync,
@@ -263,8 +264,7 @@ export async function deliverDiscordReply(params: {
   const chunkLimit = Math.min(params.textLimit, 2000);
   const replyTo = params.replyToId?.trim() || undefined;
   const replyToMode = params.replyToMode ?? "all";
-  // replyToMode=first should only apply to the first physical send.
-  const replyOnce = replyToMode === "first" || replyToMode === "batched";
+  const replyOnce = isSingleUseReplyToMode(replyToMode);
   let replyUsed = false;
   const resolveReplyTo = () => {
     if (!replyTo) {
diff --git a/extensions/discord/src/test-support/component-runtime.ts b/extensions/discord/src/test-support/component-runtime.ts
@@ -1,3 +1,4 @@
+import { isSingleUseReplyToMode } from "openclaw/plugin-sdk/reply-reference";
 import { vi, type Mock } from "vitest";
 import { parsePluginBindingApprovalCustomId } from "../../../../src/plugins/conversation-binding.js";
 import { resolvePinnedMainDmOwnerFromAllowlist } from "../../../../src/security/dm-policy-shared.js";
@@ -94,10 +95,7 @@ vi.mock("../monitor/agent-components.runtime.js", () => {
             if (params.replyToMode === "off") {
               return undefined;
             }
-            if (
-              (params.replyToMode === "first" || params.replyToMode === "batched") &&
-              hasReplied
-            ) {
+            if (isSingleUseReplyToMode(params.replyToMode ?? "off") && hasReplied) {
               return undefined;
             }
             const value = nextId;
diff --git a/extensions/matrix/src/types.ts b/extensions/matrix/src/types.ts
@@ -142,7 +142,7 @@ export type MatrixConfig = {
   blockStreaming?: boolean;
   /** Allowlist for group senders (matrix user IDs). */
   groupAllowFrom?: Array<string | number>;
-  /** Control reply threading when reply tags are present (off|first|all). */
+  /** Control reply threading when reply tags are present (off|first|all|batched). */
   replyToMode?: ReplyToMode;
   /** How to handle thread replies (off|inbound|always). */
   threadReplies?: "off" | "inbound" | "always";
diff --git a/extensions/slack/src/action-runtime.ts b/extensions/slack/src/action-runtime.ts
@@ -1,4 +1,5 @@
 import type { AgentToolResult } from "@mariozechner/pi-agent-core";
+import { isSingleUseReplyToMode } from "openclaw/plugin-sdk/reply-reference";
 import { parseSlackBlocksInput } from "./blocks-input.js";
 import {
   createActionGate,
@@ -77,7 +78,7 @@ export type SlackActionContext = {
   currentThreadTs?: string;
   /** Reply-to mode for auto-threading. */
   replyToMode?: "off" | "first" | "all" | "batched";
-  /** Mutable ref to track if a reply was sent (for "first" mode). */
+  /** Mutable ref to track if a reply was sent for single-use reply modes. */
   hasRepliedRef?: { value: boolean };
   /** Allowed local media directories for file uploads. */
   mediaLocalRoots?: readonly string[];
@@ -87,7 +88,7 @@ export type SlackActionContext = {
 /**
  * Resolve threadTs for a Slack message based on context and replyToMode.
  * - "all": always inject threadTs
- * - "first": inject only for first message (updates hasRepliedRef)
+ * - "first"/"batched": inject only for the first eligible message (updates hasRepliedRef)
  * - "off": never auto-inject
  */
 function resolveThreadTsFromContext(
@@ -122,7 +123,7 @@ function resolveThreadTsFromContext(
     return context.currentThreadTs;
   }
   if (
-    (context.replyToMode === "first" || context.replyToMode === "batched") &&
+    isSingleUseReplyToMode(context.replyToMode ?? "off") &&
     context.hasRepliedRef &&
     !context.hasRepliedRef.value
   ) {
diff --git a/extensions/slack/src/action-threading.ts b/extensions/slack/src/action-threading.ts
@@ -1,3 +1,4 @@
+import { isSingleUseReplyToMode } from "openclaw/plugin-sdk/reply-reference";
 import { parseSlackTarget } from "./targets.js";
 
 export function resolveSlackAutoThreadId(params: {
@@ -13,11 +14,7 @@ export function resolveSlackAutoThreadId(params: {
   if (!context?.currentThreadTs || !context.currentChannelId) {
     return undefined;
   }
-  if (
-    context.replyToMode !== "all" &&
-    context.replyToMode !== "first" &&
-    context.replyToMode !== "batched"
-  ) {
+  if (context.replyToMode !== "all" && !isSingleUseReplyToMode(context.replyToMode ?? "off")) {
     return undefined;
   }
   const parsedTarget = parseSlackTarget(params.to, { defaultKind: "channel" });
@@ -27,10 +24,7 @@ export function resolveSlackAutoThreadId(params: {
   if (parsedTarget.id.toLowerCase() !== context.currentChannelId.toLowerCase()) {
     return undefined;
   }
-  if (
-    (context.replyToMode === "first" || context.replyToMode === "batched") &&
-    context.hasRepliedRef?.value
-  ) {
+  if (isSingleUseReplyToMode(context.replyToMode ?? "off") && context.hasRepliedRef?.value) {
     return undefined;
   }
   return context.currentThreadTs;
diff --git a/src/auto-reply/reply/agent-runner-payloads.ts b/src/auto-reply/reply/agent-runner-payloads.ts
@@ -5,7 +5,7 @@ import { logVerbose } from "../../globals.js";
 import { stripHeartbeatToken } from "../heartbeat.js";
 import type { OriginatingChannelType } from "../templating.js";
 import { SILENT_REPLY_TOKEN } from "../tokens.js";
-import type { ReplyPayload } from "../types.js";
+import type { ReplyPayload, ReplyThreadingPolicy } from "../types.js";
 import { formatBunFetchSocketError, isBunFetchSocketError } from "./agent-runner-utils.js";
 import { createBlockReplyContentKey, type BlockReplyPipeline } from "./block-reply-pipeline.js";
 import {
@@ -99,7 +99,7 @@ export async function buildReplyPayloads(params: {
   replyToMode: ReplyToMode;
   replyToChannel?: OriginatingChannelType;
   currentMessageId?: string;
-  allowImplicitReplyToCurrentMessage?: boolean;
+  replyThreading?: ReplyThreadingPolicy;
   messageProvider?: string;
   messagingToolSentTexts?: string[];
   messagingToolSentMediaUrls?: string[];
@@ -141,7 +141,7 @@ export async function buildReplyPayloads(params: {
         replyToMode: params.replyToMode,
         replyToChannel: params.replyToChannel,
         currentMessageId: params.currentMessageId,
-        allowImplicitReplyToCurrentMessage: params.allowImplicitReplyToCurrentMessage,
+        replyThreading: params.replyThreading,
       }).map(async (payload) => {
         const parsed = normalizeReplyPayloadDirectives({
           payload,
diff --git a/src/auto-reply/reply/agent-runner.ts b/src/auto-reply/reply/agent-runner.ts
@@ -595,7 +595,7 @@ export async function runReplyAgent(params: {
       replyToMode,
       replyToChannel,
       currentMessageId: sessionCtx.MessageSidFull ?? sessionCtx.MessageSid,
-      allowImplicitReplyToCurrentMessage: sessionCtx.AllowImplicitReplyToCurrentMessage,
+      replyThreading: sessionCtx.ReplyThreading,
       messageProvider: followupRun.run.messageProvider,
       messagingToolSentTexts: runResult.messagingToolSentTexts,
       messagingToolSentMediaUrls: runResult.messagingToolSentMediaUrls,
diff --git a/src/auto-reply/reply/reply-payloads-base.ts b/src/auto-reply/reply/reply-payloads-base.ts
@@ -1,9 +1,12 @@
 import type { ReplyToMode } from "../../config/types.js";
 import { hasReplyPayloadContent } from "../../interactive/payload.js";
 import type { OriginatingChannelType } from "../templating.js";
-import type { ReplyPayload } from "../types.js";
+import type { ReplyPayload, ReplyThreadingPolicy } from "../types.js";
 import { extractReplyToTag } from "./reply-tags.js";
-import { createReplyToModeFilterForChannel } from "./reply-threading.js";
+import {
+  createReplyToModeFilterForChannel,
+  resolveImplicitCurrentMessageReplyAllowance,
+} from "./reply-threading.js";
 
 export function formatBtwTextForExternalDelivery(payload: ReplyPayload): string | undefined {
   const text = payload.text?.trim();
@@ -23,14 +26,14 @@ function resolveReplyThreadingForPayload(params: {
   replyToMode?: ReplyToMode;
   implicitReplyToId?: string;
   currentMessageId?: string;
-  allowImplicitReplyToCurrentMessage?: boolean;
+  replyThreading?: ReplyThreadingPolicy;
 }): ReplyPayload {
   const implicitReplyToId = params.implicitReplyToId?.trim() || undefined;
   const currentMessageId = params.currentMessageId?.trim() || undefined;
-  const allowImplicitReplyToCurrentMessage =
-    params.replyToMode === "batched"
-      ? params.allowImplicitReplyToCurrentMessage === true
-      : params.allowImplicitReplyToCurrentMessage !== false;
+  const allowImplicitReplyToCurrentMessage = resolveImplicitCurrentMessageReplyAllowance(
+    params.replyToMode,
+    params.replyThreading,
+  );
 
   let resolved: ReplyPayload =
     params.payload.replyToId ||
@@ -84,15 +87,9 @@ export function applyReplyThreading(params: {
   replyToMode: ReplyToMode;
   replyToChannel?: OriginatingChannelType;
   currentMessageId?: string;
-  allowImplicitReplyToCurrentMessage?: boolean;
+  replyThreading?: ReplyThreadingPolicy;
 }): ReplyPayload[] {
-  const {
-    payloads,
-    replyToMode,
-    replyToChannel,
-    currentMessageId,
-    allowImplicitReplyToCurrentMessage,
-  } = params;
+  const { payloads, replyToMode, replyToChannel, currentMessageId, replyThreading } = params;
   const applyReplyToMode = createReplyToModeFilterForChannel(replyToMode, replyToChannel);
   const implicitReplyToId = currentMessageId?.trim() || undefined;
   return payloads
@@ -102,7 +99,7 @@ export function applyReplyThreading(params: {
         replyToMode,
         implicitReplyToId,
         currentMessageId,
-        allowImplicitReplyToCurrentMessage,
+        replyThreading,
       }),
     )
     .filter(isRenderablePayload)
diff --git a/src/auto-reply/reply/reply-plumbing.test.ts b/src/auto-reply/reply/reply-plumbing.test.ts
@@ -224,7 +224,7 @@ describe("applyReplyThreading auto-threading", () => {
       payloads: [{ text: "A" }, { text: "B" }],
       replyToMode: "batched",
       currentMessageId: "42",
-      allowImplicitReplyToCurrentMessage: true,
+      replyThreading: { implicitCurrentMessage: "allow" },
     });
 
     expect(result).toHaveLength(2);
@@ -237,7 +237,7 @@ describe("applyReplyThreading auto-threading", () => {
       payloads: [{ text: "Hello" }],
       replyToMode: "batched",
       currentMessageId: "42",
-      allowImplicitReplyToCurrentMessage: false,
+      replyThreading: { implicitCurrentMessage: "deny" },
     });
 
     expect(result).toHaveLength(1);
@@ -249,7 +249,7 @@ describe("applyReplyThreading auto-threading", () => {
       payloads: [{ text: "Hello [[reply_to_current]]" }],
       replyToMode: "batched",
       currentMessageId: "42",
-      allowImplicitReplyToCurrentMessage: false,
+      replyThreading: { implicitCurrentMessage: "deny" },
     });
 
     expect(result).toHaveLength(1);
@@ -300,7 +300,7 @@ describe("applyReplyThreading auto-threading", () => {
     });
 
     expect(result).toHaveLength(1);
-    expect(result[0].replyToId).toBeUndefined();
+    expect(result[0].replyToId).toBe("42");
     expect(result[0].replyToTag).toBe(true);
   });
 
@@ -313,7 +313,7 @@ describe("applyReplyThreading auto-threading", () => {
     });
 
     expect(result).toHaveLength(1);
-    expect(result[0].replyToId).toBeUndefined();
+    expect(result[0].replyToId).toBe("42");
     expect(result[0].replyToTag).toBe(true);
   });
 
diff --git a/src/auto-reply/reply/reply-reference.ts b/src/auto-reply/reply/reply-reference.ts
@@ -9,6 +9,10 @@ export type ReplyReferencePlanner = {
   hasReplied(): boolean;
 };
 
+export function isSingleUseReplyToMode(mode: ReplyToMode): boolean {
+  return mode === "first" || mode === "batched";
+}
+
 export function createReplyReferencePlanner(options: {
   replyToMode: ReplyToMode;
   /** Existing thread/reference id (preferred when allowed by replyToMode). */
@@ -40,12 +44,11 @@ export function createReplyReferencePlanner(options: {
       hasReplied = true;
       return id;
     }
-    // "first" and "batched": only the first eligible reply gets a reference.
-    if (!hasReplied) {
-      hasReplied = true;
-      return id;
+    if (isSingleUseReplyToMode(options.replyToMode) && hasReplied) {
+      return undefined;
     }
-    return undefined;
+    hasReplied = true;
+    return id;
   };
 
   const markSent = () => {
diff --git a/src/auto-reply/reply/reply-threading.ts b/src/auto-reply/reply/reply-threading.ts
diff --git a/src/auto-reply/reply/reply-utils.test.ts b/src/auto-reply/reply/reply-utils.test.ts
diff --git a/src/auto-reply/templating.ts b/src/auto-reply/templating.ts
diff --git a/src/auto-reply/types.ts b/src/auto-reply/types.ts
diff --git a/src/config/types.googlechat.ts b/src/config/types.googlechat.ts
diff --git a/src/config/types.slack.ts b/src/config/types.slack.ts
diff --git a/src/config/types.telegram.ts b/src/config/types.telegram.ts
diff --git a/src/plugin-sdk/reply-reference.ts b/src/plugin-sdk/reply-reference.ts
