refactor(reply): type reply threading policy

This commit is contained in:
Peter Steinberger 2026-04-05 21:40:46 +01:00
parent 456ad889c7
commit 9b7002ee59
No known key found for this signature in database
24 changed files with 128 additions and 86 deletions

View File

@ -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

View File

@ -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`

View File

@ -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,

View File

@ -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" });
});
});

View File

@ -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;
}

View File

@ -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) {

View File

@ -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) {

View File

@ -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;

View File

@ -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";

View File

@ -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
) {

View File

@ -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;

View File

@ -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,

View File

@ -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,

View File

@ -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)

View File

@ -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);
});

View File

@ -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 = () => {

View File

@ -7,7 +7,8 @@ import { normalizeChannelId as normalizeBuiltInChannelId } from "../../channels/
import type { OpenClawConfig } from "../../config/config.js";
import type { ReplyToMode } from "../../config/types.js";
import type { OriginatingChannelType } from "../templating.js";
import type { ReplyPayload } from "../types.js";
import type { ReplyPayload, ReplyThreadingPolicy } from "../types.js";
import { isSingleUseReplyToMode } from "./reply-reference.js";
type ReplyToModeChannelConfig = {
replyToMode?: ReplyToMode;
@ -124,8 +125,7 @@ export function createReplyToModeFilter(
if (mode === "all") {
return payload;
}
// "first" and "batched" both keep only the first eligible physical send.
if (hasThreaded) {
if (isSingleUseReplyToMode(mode) && hasThreaded) {
// Compaction notices are transient status messages that should always
// appear in-thread, even after the first assistant block has already
// consumed the "first" slot. Let them keep their replyToId.
@ -138,13 +138,39 @@ export function createReplyToModeFilter(
// threaded (so they appear in-context), but they must not consume the
// "first" slot of the replyToMode=first|batched filter. Skip advancing
// hasThreaded so the real assistant reply still gets replyToId.
if (!payload.isCompactionNotice) {
if (isSingleUseReplyToMode(mode) && !payload.isCompactionNotice) {
hasThreaded = true;
}
return payload;
};
}
export function resolveImplicitCurrentMessageReplyAllowance(
mode: ReplyToMode | undefined,
policy?: ReplyThreadingPolicy,
): boolean {
const implicitCurrentMessage = policy?.implicitCurrentMessage ?? "default";
if (implicitCurrentMessage === "allow") {
return true;
}
if (implicitCurrentMessage === "deny") {
return false;
}
return mode !== "batched";
}
export function resolveBatchedReplyThreadingPolicy(
mode: ReplyToMode,
isBatched: boolean,
): ReplyThreadingPolicy | undefined {
if (mode !== "batched") {
return undefined;
}
return {
implicitCurrentMessage: isBatched ? "allow" : "deny",
};
}
export function createReplyToModeFilterForChannel(
mode: ReplyToMode,
channel?: OriginatingChannelType,

View File

@ -4,7 +4,7 @@ import { parseAudioTag } from "./audio-tags.js";
import { createBlockReplyCoalescer } from "./block-reply-coalescer.js";
import { matchesMentionWithExplicit } from "./mentions.js";
import { normalizeReplyPayload } from "./normalize-reply.js";
import { createReplyReferencePlanner } from "./reply-reference.js";
import { createReplyReferencePlanner, isSingleUseReplyToMode } from "./reply-reference.js";
import {
extractShortModelName,
hasTemplateVariables,
@ -888,6 +888,13 @@ describe("createReplyReferencePlanner", () => {
});
expect(existingIdPlanner.use()).toBe("thread-1");
expect(existingIdPlanner.use()).toBeUndefined();
const batchedPlanner = createReplyReferencePlanner({
replyToMode: "batched",
startId: "parent",
});
expect(batchedPlanner.use()).toBe("parent");
expect(batchedPlanner.use()).toBeUndefined();
});
it("honors allowReference=false", () => {
@ -903,6 +910,15 @@ describe("createReplyReferencePlanner", () => {
});
});
describe("isSingleUseReplyToMode", () => {
it("treats first and batched as single-use reply modes", () => {
expect(isSingleUseReplyToMode("off")).toBe(false);
expect(isSingleUseReplyToMode("all")).toBe(false);
expect(isSingleUseReplyToMode("first")).toBe(true);
expect(isSingleUseReplyToMode("batched")).toBe(true);
});
});
describe("createStreamingDirectiveAccumulator", () => {
it("stashes reply_to_current until a renderable chunk arrives", () => {
const accumulator = createStreamingDirectiveAccumulator();

View File

@ -5,6 +5,7 @@ import type {
} from "../media-understanding/types.js";
import type { InputProvenance } from "../sessions/input-provenance.js";
import type { CommandArgs } from "./commands-registry.types.js";
import type { ReplyThreadingPolicy } from "./types.js";
/** Valid message channels for routing. */
export type OriginatingChannelType = ChannelId;
@ -64,11 +65,8 @@ export type MsgContext = {
MessageSids?: string[];
MessageSidFirst?: string;
MessageSidLast?: string;
/**
* Whether this inbound turn should implicitly reply to the current message
* when reply threading is enabled. Undefined preserves legacy behavior.
*/
AllowImplicitReplyToCurrentMessage?: boolean;
/** Per-turn reply-threading overrides. */
ReplyThreading?: ReplyThreadingPolicy;
ReplyToId?: string;
/**
* Root message id for thread reconstruction (used by Feishu for root_id).

View File

@ -24,6 +24,11 @@ export type TypingPolicy =
| "internal_webchat"
| "heartbeat";
export type ReplyThreadingPolicy = {
/** Override implicit reply-to-current behavior for the current turn. */
implicitCurrentMessage?: "default" | "allow" | "deny";
};
export type GetReplyOptions = {
/** Override run id for agent events (defaults to random UUID). */
runId?: string;

View File

@ -95,7 +95,7 @@ export type GoogleChatAccountConfig = {
/** Merge streamed block replies before sending. */
blockStreamingCoalesce?: BlockStreamingCoalesceConfig;
mediaMaxMb?: 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;
/** Per-action tool gating (default: true for all). */
actions?: GoogleChatActionConfig;

View File

@ -172,7 +172,7 @@ export type SlackAccountConfig = {
reactionNotifications?: SlackReactionNotificationMode;
/** Allowlist for reaction notifications when mode is allowlist. */
reactionAllowlist?: 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;
/**
* Optional per-chat-type reply threading overrides.

View File

@ -120,7 +120,7 @@ export type TelegramAccountConfig = {
botToken?: string;
/** Path to a regular file containing the bot token; symlinks are rejected. */
tokenFile?: string;
/** 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;
groups?: Record<string, TelegramGroupConfig>;
/** Per-DM configuration for Telegram DM topics (key is chat ID). */

View File

@ -1 +1,6 @@
export { createReplyReferencePlanner } from "../auto-reply/reply/reply-reference.js";
export {
createReplyReferencePlanner,
isSingleUseReplyToMode,
} from "../auto-reply/reply/reply-reference.js";
export { resolveBatchedReplyThreadingPolicy } from "../auto-reply/reply/reply-threading.js";
export type { ReplyThreadingPolicy } from "../auto-reply/types.js";